taolib.testing.email_service.providers

taolib.testing.email_service.providers#

邮件提供商层。

Submodules#

Classes#

`ProviderFailoverManager`	提供商故障转移管理器。
`MailgunProvider`	Mailgun 邮件提供商。
`EmailProviderProtocol`	邮件提供商协议。
`ProviderHealthStatus`	提供商健康状态。
`SendResult`	提供商发送结果。
`SendGridProvider`	SendGrid 邮件提供商。
`SESProvider`	Amazon SES 邮件提供商。

Package Contents#

class taolib.testing.email_service.providers.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.providers.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.providers.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.providers.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.providers.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.providers.SendGridProvider(api_key: str, sender_email: str | None = None, sender_name: str | None = None)#