taolib.testing.email_service

邮件服务系统。

提供完整的事务性邮件和营销邮件发送、模板管理、投递追踪和退订管理功能。 支持多个邮件服务提供商（SendGrid、Mailgun、Amazon SES、SMTP）自动故障转移。

Submodules

• taolib.testing.email_service.cli
• taolib.testing.email_service.errors
• taolib.testing.email_service.events
• taolib.testing.email_service.models
• taolib.testing.email_service.providers
• taolib.testing.email_service.queue
• taolib.testing.email_service.repository
• taolib.testing.email_service.server
• taolib.testing.email_service.services
• taolib.testing.email_service.template

Exceptions

AllProvidersFailedError	所有提供商均发送失败。
EmailNotFoundError	邮件记录未找到。
EmailServiceError	邮件服务基础异常。
ProviderError	邮件提供商通信错误。
QueueError	队列操作错误。
SubscriptionError	订阅操作错误。
TemplateNotFoundError	模板未找到。
TemplateRenderError	模板渲染失败。

Classes

BounceType	Bounce classification.
EmailAttachment	Email attachment metadata.
EmailCreate	Email creation request.
EmailDocument	Email MongoDB document.
EmailPriority	Email sending priority.
EmailRecipient	Email recipient with optional display name.
EmailResponse	Email API response.
EmailStatus	Email delivery status.
EmailType	Email category type.
ProviderType	Email service provider type.
SubscriptionDocument	Subscription MongoDB document.
SubscriptionResponse	Subscription API response.
SubscriptionStatus	Subscription status.
TemplateCreate	Template creation request.
TemplateDocument	Template MongoDB document.
TemplateResponse	Template API response.
TemplateUpdate	Template update request (all fields optional).
TrackingEventDocument	Tracking event MongoDB document.
TrackingEventResponse	Tracking event API response.
TrackingEventType	Tracking event type for email analytics.
EmailProviderProtocol	邮件提供商协议。
MailgunProvider	Mailgun 邮件提供商。
ProviderFailoverManager	提供商故障转移管理器。
ProviderHealthStatus	提供商健康状态。
SendGridProvider	SendGrid 邮件提供商。
SendResult	提供商发送结果。
SESProvider	Amazon SES 邮件提供商。
EmailQueueProtocol	邮件队列协议。
InMemoryEmailQueue	内存邮件队列。
RedisEmailQueue	Redis 邮件队列。
BounceHandler	退信处理器。
EmailAnalytics	邮件分析数据。
EmailService	核心邮件发送服务。
QueueProcessor	邮件队列处理器。
SubscriptionService	订阅管理服务。
TemplateService	模板管理服务。
TrackingService	追踪和分析服务。
RenderedEmail	渲染后的邮件内容。
TemplateEngine	Jinja2 沙箱模板引擎。

Package Contents

exception taolib.testing.email_service.AllProvidersFailedError(errors: list[tuple[str, str]] | None = None)

Bases: ProviderError

所有提供商均发送失败。

errors = []

exception taolib.testing.email_service.EmailNotFoundError

Bases: EmailServiceError

邮件记录未找到。

exception taolib.testing.email_service.EmailServiceError

Bases: Exception

邮件服务基础异常。

exception taolib.testing.email_service.ProviderError

Bases: EmailServiceError

邮件提供商通信错误。

exception taolib.testing.email_service.QueueError

Bases: EmailServiceError

队列操作错误。

exception taolib.testing.email_service.SubscriptionError

Bases: EmailServiceError

订阅操作错误。

exception taolib.testing.email_service.TemplateNotFoundError

Bases: EmailServiceError

模板未找到。

exception taolib.testing.email_service.TemplateRenderError

Bases: EmailServiceError

模板渲染失败。

class taolib.testing.email_service.BounceType

Bases: enum.StrEnum

Bounce classification.

HARD = 'hard'

SOFT = 'soft'

UNDETERMINED = 'undetermined'

class taolib.testing.email_service.EmailAttachment

Bases: pydantic.BaseModel

Email attachment metadata.

filename: str

content_type: str

content_base64: str

class taolib.testing.email_service.EmailCreate

Bases: EmailBase

Email creation request.

template_id: str | None

template_variables: dict[str, str | int | float | bool | list | dict]

html_body: str | None

text_body: str | None

attachments: list[EmailAttachment]

schedule_at: datetime.datetime | None

metadata: dict[str, str]

class taolib.testing.email_service.EmailDocument

Bases: EmailBase

Email MongoDB document.

id: str

status: taolib.testing.email_service.models.enums.EmailStatus

provider: str | None

provider_message_id: str | None

template_id: str | None

html_body: str | None

text_body: str | None

attachments: list[EmailAttachment]

retry_count: int

max_retries: int

error_message: str | None

schedule_at: datetime.datetime | None

sent_at: datetime.datetime | None

delivered_at: datetime.datetime | None

opened_at: datetime.datetime | None

created_at: datetime.datetime

updated_at: datetime.datetime

created_by: str

metadata: dict[str, str]

model_config

to_response() → EmailResponse

Convert to API response model.

class taolib.testing.email_service.EmailPriority

Bases: enum.StrEnum

Email sending priority.

HIGH = 'high'

NORMAL = 'normal'

LOW = 'low'

class taolib.testing.email_service.EmailRecipient

Bases: pydantic.BaseModel

Email recipient with optional display name.

email: str

name: str | None

class taolib.testing.email_service.EmailResponse

Bases: EmailBase

Email API response.

id: str

status: taolib.testing.email_service.models.enums.EmailStatus

provider: str | None

provider_message_id: str | None

template_id: str | None

html_body: str | None

text_body: str | None

retry_count: int

error_message: str | None

sent_at: datetime.datetime | None

delivered_at: datetime.datetime | None

opened_at: datetime.datetime | None

created_at: datetime.datetime

updated_at: datetime.datetime

metadata: dict[str, str]

model_config

class taolib.testing.email_service.EmailStatus

Bases: enum.StrEnum

Email delivery status.

QUEUED = 'queued'

SENDING = 'sending'

SENT = 'sent'

DELIVERED = 'delivered'

OPENED = 'opened'

CLICKED = 'clicked'

BOUNCED = 'bounced'

FAILED = 'failed'

REJECTED = 'rejected'

class taolib.testing.email_service.EmailType

Bases: enum.StrEnum

Email category type.

TRANSACTIONAL = 'transactional'

MARKETING = 'marketing'

class taolib.testing.email_service.ProviderType

Bases: enum.StrEnum

Email service provider type.

SENDGRID = 'sendgrid'

MAILGUN = 'mailgun'

SES = 'ses'

SMTP = 'smtp'

class taolib.testing.email_service.SubscriptionDocument

Bases: pydantic.BaseModel

Subscription MongoDB document.

Tracks whether a recipient has unsubscribed from marketing emails.

id: str

email: str

status: taolib.testing.email_service.models.enums.SubscriptionStatus

unsubscribe_reason: str | None

unsubscribe_token: str

tags: list[str]

subscribed_at: datetime.datetime

unsubscribed_at: datetime.datetime | None

created_at: datetime.datetime

updated_at: datetime.datetime

model_config

to_response() → SubscriptionResponse

Convert to API response model.

class taolib.testing.email_service.SubscriptionResponse

Bases: pydantic.BaseModel

Subscription API response.

id: str

email: str

status: taolib.testing.email_service.models.enums.SubscriptionStatus

unsubscribe_reason: str | None = None

tags: list[str]

subscribed_at: datetime.datetime

unsubscribed_at: datetime.datetime | None = None

created_at: datetime.datetime

updated_at: datetime.datetime

model_config

class taolib.testing.email_service.SubscriptionStatus

Bases: enum.StrEnum

Subscription status.

ACTIVE = 'active'

UNSUBSCRIBED = 'unsubscribed'

class taolib.testing.email_service.TemplateCreate

Bases: TemplateBase

Template creation request.

class taolib.testing.email_service.TemplateDocument

Bases: TemplateBase

Template MongoDB document.

id: str

is_active: bool

version: int

created_at: datetime.datetime

updated_at: datetime.datetime

created_by: str

model_config

to_response() → TemplateResponse

Convert to API response model.

class taolib.testing.email_service.TemplateResponse

Bases: TemplateBase

Template API response.

id: str

is_active: bool

version: int

created_at: datetime.datetime

updated_at: datetime.datetime

model_config

class taolib.testing.email_service.TemplateUpdate

Bases: pydantic.BaseModel

Template update request (all fields optional).

name: str | None = None

description: str | None = None

subject_template: str | None = None

html_template: str | None = None

text_template: str | None = None

email_type: taolib.testing.email_service.models.enums.EmailType | None = None

variables_schema: dict[str, str] | None = None

tags: list[str] | None = None

is_active: bool | None = None

class taolib.testing.email_service.TrackingEventDocument

Bases: TrackingEventBase

Tracking event MongoDB document.

id: str

provider: str | None

ip_address: str | None

user_agent: str | None

click_url: str | None

bounce_type: taolib.testing.email_service.models.enums.BounceType | None

bounce_reason: str | None

raw_payload: dict | None

created_at: datetime.datetime

model_config

to_response() → TrackingEventResponse

Convert to API response model.

class taolib.testing.email_service.TrackingEventResponse

Bases: TrackingEventBase

Tracking event API response.

id: str

provider: str | None

ip_address: str | None

user_agent: str | None

click_url: str | None

bounce_type: taolib.testing.email_service.models.enums.BounceType | None

bounce_reason: str | None

created_at: datetime.datetime

model_config

class taolib.testing.email_service.TrackingEventType

Bases: enum.StrEnum

Tracking event type for email analytics.

SENT = 'sent'

DELIVERED = 'delivered'

OPENED = 'opened'

CLICKED = 'clicked'

BOUNCED = 'bounced'

COMPLAINED = 'complained'

UNSUBSCRIBED = 'unsubscribed'

class taolib.testing.email_service.EmailProviderProtocol

Bases: Protocol

邮件提供商协议。

所有邮件服务提供商必须实现此协议。

property name: str

提供商名称。

async send(email: taolib.testing.email_service.models.email.EmailDocument) → SendResult

发送单封邮件。

参数:

email -- 邮件文档

返回:

发送结果

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailDocument]) → list[SendResult]

批量发送邮件。

参数:

emails -- 邮件文档列表

返回:

发送结果列表

async check_health() → ProviderHealthStatus

检查提供商健康状态。

返回:

健康状态

class taolib.testing.email_service.MailgunProvider(api_key: str, domain: str)

Mailgun 邮件提供商。

通过 Mailgun HTTP API 发送邮件。

_api_key

_domain

_client

property name: str

提供商名称。

async send(email: taolib.testing.email_service.models.email.EmailDocument) → taolib.testing.email_service.providers.protocol.SendResult

通过 Mailgun API 发送邮件。

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailDocument]) → list[taolib.testing.email_service.providers.protocol.SendResult]

批量发送（逐封发送）。

async check_health() → taolib.testing.email_service.providers.protocol.ProviderHealthStatus

检查 Mailgun 可用性。

_build_form_data(email: taolib.testing.email_service.models.email.EmailDocument) → dict

构建 Mailgun 表单数据。

class taolib.testing.email_service.ProviderFailoverManager(providers: list[tuple[taolib.testing.email_service.providers.protocol.EmailProviderProtocol, int]], max_consecutive_failures: int = 3, cooldown_seconds: int = 60)

提供商故障转移管理器。

按优先级排序提供商，自动在主提供商失败时切换到备用提供商。 每个提供商独立跟踪健康状态，失败超过阈值后进入冷却期。

_states: list[_ProviderState]

_max_failures = 3

_cooldown_seconds = 60

async send(email: taolib.testing.email_service.models.email.EmailDocument) → taolib.testing.email_service.providers.protocol.SendResult

通过可用提供商发送邮件，自动故障转移。

参数:

email -- 邮件文档

返回:

发送结果

抛出:

AllProvidersFailedError -- 所有提供商均失败

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailDocument]) → list[taolib.testing.email_service.providers.protocol.SendResult]

批量发送邮件。

参数:

emails -- 邮件文档列表

返回:

发送结果列表

async run_health_checks() → None

运行健康检查，尝试恢复冷却中的提供商。

get_provider_statuses() → list[taolib.testing.email_service.providers.protocol.ProviderHealthStatus]

获取所有提供商状态。

返回:

提供商健康状态列表

class taolib.testing.email_service.ProviderHealthStatus

提供商健康状态。

provider_name: str

is_healthy: bool

consecutive_failures: int = 0

last_check: datetime.datetime | None = None

last_error: str | None = None

class taolib.testing.email_service.SendGridProvider(api_key: str, sender_email: str | None = None, sender_name: str | None = None)

SendGrid 邮件提供商。

通过 SendGrid v3 Mail Send API 发送邮件。

_api_key

_sender_email = None

_sender_name = None

_client

property name: str

提供商名称。

async send(email: taolib.testing.email_service.models.email.EmailDocument) → taolib.testing.email_service.providers.protocol.SendResult

通过 SendGrid API 发送邮件。

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailDocument]) → list[taolib.testing.email_service.providers.protocol.SendResult]

批量发送（逐封发送）。

async check_health() → taolib.testing.email_service.providers.protocol.ProviderHealthStatus

检查 SendGrid 可用性。

_build_payload(email: taolib.testing.email_service.models.email.EmailDocument) → dict

构建 SendGrid v3 API 请求体。

class taolib.testing.email_service.SendResult

提供商发送结果。

success: bool

provider_name: str

provider_message_id: str | None = None

error_message: str | None = None

latency_ms: float = 0.0

class taolib.testing.email_service.SESProvider(region: str, access_key_id: str, secret_access_key: str)

Amazon SES 邮件提供商。

通过 Amazon SES v2 HTTP API 发送邮件。 需要配置 AWS 凭证和区域。

_region

_access_key_id

_secret_access_key

_endpoint

_client

property name: str

提供商名称。

async send(email: taolib.testing.email_service.models.email.EmailDocument) → taolib.testing.email_service.providers.protocol.SendResult

通过 SES API 发送邮件。

使用简化的 SES v2 SendEmail API。 生产环境建议使用 aiobotocore 以获得完整的 AWS 签名支持。

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailDocument]) → list[taolib.testing.email_service.providers.protocol.SendResult]

批量发送（逐封发送）。

async check_health() → taolib.testing.email_service.providers.protocol.ProviderHealthStatus

检查 SES 可用性。

_build_payload(email: taolib.testing.email_service.models.email.EmailDocument) → dict

构建 SES v2 请求体。

class taolib.testing.email_service.EmailQueueProtocol

Bases: Protocol

邮件队列协议。

所有队列实现必须符合此协议。

async enqueue(email_id: str, priority: taolib.testing.email_service.models.enums.EmailPriority = EmailPriority.NORMAL) → None

将邮件 ID 加入队列。

参数:

• email_id -- 邮件 ID

• priority -- 优先级

async enqueue_bulk(email_ids: list[str], priority: taolib.testing.email_service.models.enums.EmailPriority = EmailPriority.NORMAL) → None

批量加入队列。

参数:

• email_ids -- 邮件 ID 列表

• priority -- 优先级

async dequeue(timeout: int = 5) → str | None

从队列取出一封邮件 ID。

按优先级顺序（高→普通→低）取出。

参数:

timeout -- 等待超时秒数

返回:

邮件 ID，队列为空时返回 None

async size() → dict[str, int]

获取各优先级队列的大小。

返回:

count} 字典

返回类型:

{priority

class taolib.testing.email_service.InMemoryEmailQueue

内存邮件队列。

使用 asyncio.PriorityQueue 实现，仅用于测试。

_queue: asyncio.PriorityQueue[tuple[int, str]]

_counts: dict[str, int]

async enqueue(email_id: str, priority: taolib.testing.email_service.models.enums.EmailPriority = EmailPriority.NORMAL) → None

将邮件 ID 加入队列。

async enqueue_bulk(email_ids: list[str], priority: taolib.testing.email_service.models.enums.EmailPriority = EmailPriority.NORMAL) → None

批量加入队列。

async dequeue(timeout: int = 5) → str | None

从队列取出一封邮件 ID。

async size() → dict[str, int]

获取队列大小（近似值）。

class taolib.testing.email_service.RedisEmailQueue(redis_client: redis.asyncio.Redis)

Redis 邮件队列。

使用三个 Redis List 实现优先级队列： - email:queue:high (高优先级) - email:queue:normal (普通优先级) - email:queue:low (低优先级)

BRPOP 按顺序检查三个列表，自然实现优先级。

_redis

async enqueue(email_id: str, priority: taolib.testing.email_service.models.enums.EmailPriority = EmailPriority.NORMAL) → None

将邮件 ID 加入队列。

async enqueue_bulk(email_ids: list[str], priority: taolib.testing.email_service.models.enums.EmailPriority = EmailPriority.NORMAL) → None

批量加入队列。

async dequeue(timeout: int = 5) → str | None

从队列取出一封邮件 ID（按优先级）。

使用 BRPOP 按 high → normal → low 顺序检查。

async size() → dict[str, int]

获取各优先级队列大小。

class taolib.testing.email_service.BounceHandler(tracking_service: taolib.testing.email_service.services.tracking_service.TrackingService, subscription_service: taolib.testing.email_service.services.subscription_service.SubscriptionService, email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository, hard_bounce_threshold: int = 1)

退信处理器。

_tracking

_subscription

_email_repo

_hard_bounce_threshold = 1

async handle_bounce(email_id: str, bounce_type: taolib.testing.email_service.models.enums.BounceType, reason: str, recipient: str, provider: str | None = None, raw_payload: dict | None = None) → None

处理退信事件。

参数:

• email_id -- 邮件 ID

• bounce_type -- 退信类型

• reason -- 退信原因

• recipient -- 收件人

• provider -- 提供商名称

• raw_payload -- 原始 Webhook 数据

async handle_complaint(email_id: str, recipient: str, provider: str | None = None, raw_payload: dict | None = None) → None

处理投诉事件。

投诉视为退订请求。

参数:

• email_id -- 邮件 ID

• recipient -- 收件人

• provider -- 提供商名称

• raw_payload -- 原始 Webhook 数据

class taolib.testing.email_service.EmailAnalytics

Bases: pydantic.BaseModel

邮件分析数据。

total_sent: int

total_delivered: int

total_opened: int

total_clicked: int

total_bounced: int

total_failed: int

delivery_rate: float

open_rate: float

click_rate: float

bounce_rate: float

class taolib.testing.email_service.EmailService(email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository, template_service: taolib.testing.email_service.services.template_service.TemplateService, subscription_service: taolib.testing.email_service.services.subscription_service.SubscriptionService, provider_manager: taolib.testing.email_service.providers.failover.ProviderFailoverManager, queue: taolib.testing.email_service.queue.protocol.EmailQueueProtocol, tracking_service: taolib.testing.email_service.services.tracking_service.TrackingService)

核心邮件发送服务。

编排完整的邮件发送流程： 1. 模板渲染（如有模板） 2. 订阅状态检查（营销邮件） 3. 创建邮件文档 4. 入队或直接发送

_email_repo

_template_service

_subscription_service

_provider_manager

_queue

_tracking_service

async send_email(data: taolib.testing.email_service.models.email.EmailCreate, enqueue: bool = True) → taolib.testing.email_service.models.email.EmailResponse

发送单封邮件。

参数:

• data -- 邮件创建数据

• enqueue -- 是否入队异步发送，False 则直接发送

返回:

邮件响应

async send_bulk(emails: list[taolib.testing.email_service.models.email.EmailCreate]) → list[taolib.testing.email_service.models.email.EmailResponse]

批量发送邮件。

async _send_now(email_doc: taolib.testing.email_service.models.email.EmailDocument) → taolib.testing.email_service.models.email.EmailDocument

立即发送邮件。

参数:

email_doc -- 邮件文档

返回:

更新后的邮件文档

async get_email(email_id: str) → taolib.testing.email_service.models.email.EmailResponse

获取邮件详情。

async list_emails(status: taolib.testing.email_service.models.enums.EmailStatus | None = None, email_type: taolib.testing.email_service.models.enums.EmailType | None = None, skip: int = 0, limit: int = 100) → list[taolib.testing.email_service.models.email.EmailResponse]

查询邮件列表。

class taolib.testing.email_service.QueueProcessor(queue: taolib.testing.email_service.queue.protocol.EmailQueueProtocol, email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository, send_callback, poll_interval: float = 1.0, batch_size: int = 10)

邮件队列处理器。

作为后台 asyncio.Task 运行，持续从队列取出邮件并发送。 参考 data_sync.services.scheduler.AsyncScheduler 模式。

_queue

_email_repo

_send_callback

_poll_interval = 1.0

_batch_size = 10

_running = False

_task: asyncio.Task[None] | None = None

async start() → None

启动处理器。

async stop() → None

停止处理器。

async _run_loop() → None

主处理循环。

async _process_batch() → None

处理一批邮件。

async _process_single(email_id: str) → None

处理单封邮件。

async process_scheduled() → None

处理到期的计划邮件。

class taolib.testing.email_service.SubscriptionService(subscription_repo: taolib.testing.email_service.repository.subscription_repo.SubscriptionRepository)

订阅管理服务。

_repo

async get_or_create_subscription(email: str) → taolib.testing.email_service.models.subscription.SubscriptionDocument

获取或创建订阅记录。

如果邮箱没有订阅记录，则创建一个新的激活状态记录。

参数:

email -- 邮箱地址

返回:

订阅文档

async unsubscribe(token: str, reason: str | None = None) → taolib.testing.email_service.models.subscription.SubscriptionResponse

处理退订请求。

参数:

• token -- 退订令牌

• reason -- 退订原因

返回:

更新后的订阅响应

抛出:

SubscriptionError -- 令牌无效或已退订

async unsubscribe_by_email(email: str, reason: str | None = None) → taolib.testing.email_service.models.subscription.SubscriptionResponse | None

按邮箱退订（用于硬退信自动退订）。

async resubscribe(email: str) → taolib.testing.email_service.models.subscription.SubscriptionResponse

重新订阅。

参数:

email -- 邮箱地址

返回:

更新后的订阅响应

抛出:

SubscriptionError -- 订阅记录不存在

async is_subscribed(email: str) → bool

检查是否订阅。

async get_unsubscribe_token(email: str) → str

获取退订令牌（自动创建订阅记录）。

class taolib.testing.email_service.TemplateService(template_repo: taolib.testing.email_service.repository.template_repo.TemplateRepository, engine: taolib.testing.email_service.template.engine.TemplateEngine)

模板管理服务。

_repo

_engine

async create_template(data: taolib.testing.email_service.models.template.TemplateCreate) → taolib.testing.email_service.models.template.TemplateResponse

创建模板。

async update_template(template_id: str, data: taolib.testing.email_service.models.template.TemplateUpdate) → taolib.testing.email_service.models.template.TemplateResponse | None

更新模板。

async get_template(template_id: str) → taolib.testing.email_service.models.template.TemplateResponse | None

获取模板。

async get_template_by_name(name: str) → taolib.testing.email_service.models.template.TemplateResponse | None

按名称获取模板。

async list_templates(email_type: taolib.testing.email_service.models.enums.EmailType | None = None, is_active: bool | None = None, skip: int = 0, limit: int = 100) → list[taolib.testing.email_service.models.template.TemplateResponse]

查询模板列表。

async delete_template(template_id: str) → bool

删除模板。

async render_template(template_id: str, variables: dict, email_type: taolib.testing.email_service.models.enums.EmailType = EmailType.TRANSACTIONAL, recipient_email: str | None = None, unsubscribe_token: str | None = None) → taolib.testing.email_service.template.engine.RenderedEmail

渲染模板。

参数:

• template_id -- 模板 ID

• variables -- 模板变量

• email_type -- 邮件类型

• recipient_email -- 收件人邮箱

• unsubscribe_token -- 退订令牌

返回:

渲染后的邮件内容

抛出:

TemplateNotFoundError -- 模板不存在

class taolib.testing.email_service.TrackingService(tracking_repo: taolib.testing.email_service.repository.tracking_repo.TrackingRepository, email_repo: taolib.testing.email_service.repository.email_repo.EmailRepository)

追踪和分析服务。

_tracking_repo

_email_repo

async record_event(email_id: str, event_type: taolib.testing.email_service.models.enums.TrackingEventType, recipient: str, provider: str | None = None, ip_address: str | None = None, user_agent: str | None = None, click_url: str | None = None, bounce_type: str | None = None, bounce_reason: str | None = None, raw_payload: dict | None = None) → taolib.testing.email_service.models.tracking.TrackingEventResponse

记录追踪事件。

同时更新关联的 EmailDocument 状态。

async get_events_for_email(email_id: str) → list[taolib.testing.email_service.models.tracking.TrackingEventResponse]

获取邮件的所有追踪事件。

async get_analytics(start: datetime.datetime, end: datetime.datetime) → EmailAnalytics

获取时间范围内的分析数据。

async get_daily_stats(start: datetime.datetime, end: datetime.datetime) → list[dict]

获取按日统计数据。

async _update_email_status(email_id: str, event_type: taolib.testing.email_service.models.enums.TrackingEventType, timestamp: datetime.datetime) → None

根据追踪事件更新邮件状态。

class taolib.testing.email_service.RenderedEmail

渲染后的邮件内容。

subject: str

html_body: str

text_body: str | None = None

class taolib.testing.email_service.TemplateEngine(unsubscribe_base_url: str = '')

Jinja2 沙箱模板引擎。

使用 SandboxedEnvironment 防止模板注入攻击。 StrictUndefined 确保缺失变量会立即报错。

_env: jinja2.Environment

_unsubscribe_base_url = ''

render(template_doc: taolib.testing.email_service.models.template.TemplateDocument, variables: dict, email_type: taolib.testing.email_service.models.enums.EmailType = EmailType.TRANSACTIONAL, recipient_email: str | None = None, unsubscribe_token: str | None = None) → RenderedEmail

渲染模板。

参数:

• template_doc -- 模板文档

• variables -- 模板变量

• email_type -- 邮件类型

• recipient_email -- 收件人邮箱（营销邮件需要）

• unsubscribe_token -- 退订令牌（营销邮件需要）

返回:

渲染后的邮件内容

抛出:

TemplateRenderError -- 模板渲染失败

validate_template(template_string: str, variables: dict | None = None) → list[str]

验证模板语法。

参数:

• template_string -- 模板字符串

• variables -- 测试变量

返回:

错误列表，空列表表示有效

_render_string(template_string: str, variables: dict) → str

渲染单个模板字符串。

_inject_unsubscribe_link(html_body: str, text_body: str | None, email: str, token: str) → tuple[str, str | None]

注入退订链接。