Infra
leadr.infra¶
Modules:
- blob_storage –
- cache –
- email – Email infrastructure domain with factory functions for easy integration.
leadr.infra.blob_storage¶
Modules:
leadr.infra.blob_storage.adapters¶
leadr.infra.blob_storage.domain¶
leadr.infra.blob_storage.services¶
leadr.infra.cache¶
Modules:
leadr.infra.cache.adapters¶
leadr.infra.cache.domain¶
leadr.infra.cache.services¶
leadr.infra.email¶
Email infrastructure domain with factory functions for easy integration.
Modules:
- adapters – Email adapter implementations.
- domain – Email domain models and interfaces.
- service – Email service with dependency injection and convenience methods.
Classes:
- Email – Email domain entity.
- EmailError – Base exception for email domain errors.
- EmailPriority – Email priority enumeration.
- EmailProvider – Interface for email service providers.
- EmailSendError – Raised when email sending fails.
- EmailService – Email service with dependency injection for testing and flexibility.
- EmailStatus – Email status enumeration.
- EmailValidationError – Raised when email validation fails.
- MailgunEmailProvider – Mailgun email service provider implementation.
- SMTPEmailProvider – SMTP email provider implementation for development/testing.
Functions:
- create_email_service – Create an email service with the specified or default provider.
leadr.infra.email.Email¶
Bases: Entity
Email domain entity.
Functions:
- create – Create a new Email entity.
- mark_as_delivered – Mark email as delivered.
- mark_as_failed – Mark email as failed.
- mark_as_sent – Mark email as sent.
- restore – Restore a soft-deleted entity.
- soft_delete – Mark entity as soft-deleted.
- validate_bcc_emails – Validate BCC email addresses.
- validate_body – Validate email body.
- validate_cc_emails – Validate CC email addresses.
- validate_from_email – Validate sender email address.
- validate_reply_to – Validate reply-to email address.
- validate_subject – Validate email subject.
- validate_to_email – Validate recipient email address.
Attributes:
- bcc (
list[str]) – - body (
str) – - cc (
list[str]) – - created_at (
datetime) – - deleted_at (
datetime | None) – - error_message (
str | None) – - failed_at (
datetime | None) – - from_email (
str | None) – - id (
EmailID) – - is_deleted (
bool) – Check if entity is soft-deleted. - model_config –
- priority (
EmailPriority) – - provider_message_id (
str | None) – - provider_response (
dict[str, Any] | None) – - reply_to (
str | None) – - sent_at (
datetime | None) – - status (
EmailStatus) – - subject (
str) – - template_data (
dict[str, Any] | None) – - to (
str) – - updated_at (
datetime) –
leadr.infra.email.Email.bcc¶
bcc: list[str] = Field(default_factory=list, description='BCC recipients')
leadr.infra.email.Email.body¶
body: str = Field(..., description='Email body content')
leadr.infra.email.Email.cc¶
cc: list[str] = Field(default_factory=list, description='CC recipients')
leadr.infra.email.Email.create¶
create(to, subject, body, from_email=None, reply_to=None, cc=None, bcc=None, priority=EmailPriority.NORMAL, template_data=None)
Create a new Email entity.
leadr.infra.email.Email.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
leadr.infra.email.Email.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
leadr.infra.email.Email.error_message¶
error_message: str | None = Field(None, description='Error message if failed')
leadr.infra.email.Email.failed_at¶
failed_at: datetime | None = Field(None, description='Time email failed')
leadr.infra.email.Email.from_email¶
from_email: str | None = Field(None, description='Sender email address')
leadr.infra.email.Email.id¶
id: EmailID = Field(frozen=True, default_factory=EmailID, description='Unique email identifier')
leadr.infra.email.Email.is_deleted¶
is_deleted: bool
Check if entity is soft-deleted.
Returns:
bool– True if the entity has a deleted_at timestamp, False otherwise.
leadr.infra.email.Email.mark_as_delivered¶
mark_as_delivered()
Mark email as delivered.
leadr.infra.email.Email.mark_as_failed¶
mark_as_failed(error_message, provider_response=None)
Mark email as failed.
leadr.infra.email.Email.mark_as_sent¶
mark_as_sent(provider_message_id, provider_response)
Mark email as sent.
leadr.infra.email.Email.model_config¶
model_config = ConfigDict(validate_assignment=True)
leadr.infra.email.Email.priority¶
priority: EmailPriority = Field(default=(EmailPriority.NORMAL), description='Email priority')
leadr.infra.email.Email.provider_message_id¶
provider_message_id: str | None = Field(None, description='Provider message ID')
leadr.infra.email.Email.provider_response¶
provider_response: dict[str, Any] | None = Field(None, description='Provider API response')
leadr.infra.email.Email.reply_to¶
reply_to: str | None = Field(None, description='Reply-to email address')
leadr.infra.email.Email.restore¶
restore()
Restore a soft-deleted entity.
Clears the deleted_at timestamp, making the entity active again.
Example
> > > account.soft_delete() > > > account.restore() > > > assert account.is_deleted is Falseleadr.infra.email.Email.sent_at¶
sent_at: datetime | None = Field(None, description='Time email was sent')
leadr.infra.email.Email.soft_delete¶
soft_delete()
Mark entity as soft-deleted.
Sets the deleted_at timestamp to the current UTC time. Entities that are already deleted are not affected (deleted_at remains at original deletion time).
Example
> > > account = Account(name="Test", slug="test") > > > account.soft_delete() > > > assert account.is_deleted is Trueleadr.infra.email.Email.status¶
status: EmailStatus = Field(default=(EmailStatus.PENDING), description='Email status')
leadr.infra.email.Email.subject¶
subject: str = Field(..., description='Email subject')
leadr.infra.email.Email.template_data¶
template_data: dict[str, Any] | None = Field(None, description='Template data for email rendering')
leadr.infra.email.Email.to¶
to: str = Field(..., description='Recipient email address')
leadr.infra.email.Email.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
leadr.infra.email.Email.validate_bcc_emails¶
validate_bcc_emails(v)
Validate BCC email addresses.
leadr.infra.email.Email.validate_body¶
validate_body(v)
Validate email body.
leadr.infra.email.Email.validate_cc_emails¶
validate_cc_emails(v)
Validate CC email addresses.
leadr.infra.email.Email.validate_from_email¶
validate_from_email(v)
Validate sender email address.
leadr.infra.email.Email.validate_reply_to¶
validate_reply_to(v)
Validate reply-to email address.
leadr.infra.email.Email.validate_subject¶
validate_subject(v)
Validate email subject.
leadr.infra.email.Email.validate_to_email¶
validate_to_email(v)
Validate recipient email address.
leadr.infra.email.EmailError¶
Bases: Exception
Base exception for email domain errors.
leadr.infra.email.EmailPriority¶
Email priority enumeration.
Attributes:
leadr.infra.email.EmailPriority.HIGH¶
HIGH = 'high'
leadr.infra.email.EmailPriority.LOW¶
LOW = 'low'
leadr.infra.email.EmailPriority.NORMAL¶
NORMAL = 'normal'
leadr.infra.email.EmailPriority.URGENT¶
URGENT = 'urgent'
leadr.infra.email.EmailProvider¶
Bases: ABC
Interface for email service providers.
Functions:
- send – Send an email and return provider response.
- validate_config – Validate provider configuration.
leadr.infra.email.EmailProvider.send¶
send(email)
Send an email and return provider response.
leadr.infra.email.EmailProvider.validate_config¶
validate_config()
Validate provider configuration.
leadr.infra.email.EmailSendError¶
EmailSendError(message, provider_response=None)
Bases: EmailError
Raised when email sending fails.
Attributes:
leadr.infra.email.EmailSendError.provider_response¶
provider_response = provider_response
leadr.infra.email.EmailService¶
EmailService(provider, db=None, validate_on_init=False)
Email service with dependency injection for testing and flexibility.
Functions:
- get_default_from_email – Get default from email address.
- send_email – Send an email using the configured provider.
- send_notification_email – Send a notification email.
- send_verification_code – Send a verification code email for LEADR registration.
- send_welcome_email – Send a welcome email after successful LEADR registration.
- validate_provider_config – Validate the email provider configuration.
Attributes:
- db –
- provider –
- repository –
- templates_dir –
Parameters:
- provider (
EmailProvider) – Email provider implementation (e.g., Mailgun). - db (
AsyncSession | None) – Optional database session for persisting email records. - validate_on_init (
bool) – Whether to validate provider config on initialization.
leadr.infra.email.EmailService.db¶
db = db
leadr.infra.email.EmailService.get_default_from_email¶
get_default_from_email()
Get default from email address.
leadr.infra.email.EmailService.provider¶
provider = provider
leadr.infra.email.EmailService.repository¶
repository = EmailRepository(db) if db else None
leadr.infra.email.EmailService.send_email¶
send_email(to, subject, body, from_email=None, reply_to=None, cc=None, bcc=None, priority=EmailPriority.NORMAL, template_data=None)
Send an email using the configured provider.
leadr.infra.email.EmailService.send_notification_email¶
send_notification_email(to, subject, message, priority=EmailPriority.NORMAL, from_email=None)
Send a notification email.
leadr.infra.email.EmailService.send_verification_code¶
send_verification_code(to, code)
Send a verification code email for LEADR registration.
leadr.infra.email.EmailService.send_welcome_email¶
send_welcome_email(to, user_name, account_name, account_slug, from_email=None)
Send a welcome email after successful LEADR registration.
leadr.infra.email.EmailService.templates_dir¶
templates_dir = Path(__file__).parent / 'templates'
leadr.infra.email.EmailService.validate_provider_config¶
validate_provider_config()
Validate the email provider configuration.
leadr.infra.email.EmailStatus¶
Email status enumeration.
Attributes:
leadr.infra.email.EmailStatus.DELIVERED¶
DELIVERED = 'delivered'
leadr.infra.email.EmailStatus.FAILED¶
FAILED = 'failed'
leadr.infra.email.EmailStatus.PENDING¶
PENDING = 'pending'
leadr.infra.email.EmailStatus.SENT¶
SENT = 'sent'
leadr.infra.email.EmailValidationError¶
Bases: EmailError
Raised when email validation fails.
leadr.infra.email.MailgunEmailProvider¶
MailgunEmailProvider()
Bases: EmailProvider
Mailgun email service provider implementation.
Functions:
- send – Send email via Mailgun API.
- validate_config – Validate Mailgun configuration.
Attributes:
leadr.infra.email.MailgunEmailProvider.api_key¶
api_key = settings.MAILGUN_API_KEY
leadr.infra.email.MailgunEmailProvider.client¶
client = Client(auth=('api', self.api_key))
leadr.infra.email.MailgunEmailProvider.domain¶
domain = settings.MAILGUN_DOMAIN
leadr.infra.email.MailgunEmailProvider.send¶
send(email)
Send email via Mailgun API.
leadr.infra.email.MailgunEmailProvider.validate_config¶
validate_config()
Validate Mailgun configuration.
leadr.infra.email.SMTPEmailProvider¶
SMTPEmailProvider(host=None, port=None)
Bases: EmailProvider
SMTP email provider implementation for development/testing.
Functions:
- send – Send email via SMTP.
- validate_config – Validate SMTP configuration.
Attributes:
Parameters:
- host (
str | None) – SMTP server hostname (defaults to settings.SMTP_HOST or localhost) - port (
int | None) – SMTP server port (defaults to settings.SMTP_PORT or 1025)
leadr.infra.email.SMTPEmailProvider.host¶
host = host or getattr(settings, 'SMTP_HOST', 'localhost')
leadr.infra.email.SMTPEmailProvider.port¶
port = port or getattr(settings, 'SMTP_PORT', 1025)
leadr.infra.email.SMTPEmailProvider.send¶
send(email)
Send email via SMTP.
leadr.infra.email.SMTPEmailProvider.validate_config¶
validate_config()
Validate SMTP configuration.
leadr.infra.email.adapters¶
Email adapter implementations.
Modules:
- mailgun – Mailgun email adapter implementation.
- orm – Email ORM models.
- repositories – Email repository for database operations.
- smtp – SMTP email adapter implementation for testing.
leadr.infra.email.adapters.mailgun¶
Mailgun email adapter implementation.
Classes:
- MailgunEmailProvider – Mailgun email service provider implementation.
# leadr.infra.email.adapters.mailgun.MailgunEmailProvider¶
MailgunEmailProvider()
Bases: EmailProvider
Mailgun email service provider implementation.
Functions:
- send – Send email via Mailgun API.
- validate_config – Validate Mailgun configuration.
Attributes:
## leadr.infra.email.adapters.mailgun.MailgunEmailProvider.api_key¶
api_key = settings.MAILGUN_API_KEY
## leadr.infra.email.adapters.mailgun.MailgunEmailProvider.client¶
client = Client(auth=('api', self.api_key))
## leadr.infra.email.adapters.mailgun.MailgunEmailProvider.domain¶
domain = settings.MAILGUN_DOMAIN
## leadr.infra.email.adapters.mailgun.MailgunEmailProvider.send¶
send(email)
Send email via Mailgun API.
## leadr.infra.email.adapters.mailgun.MailgunEmailProvider.validate_config¶
validate_config()
Validate Mailgun configuration.
leadr.infra.email.adapters.orm¶
Email ORM models.
Classes:
- EmailORM – Email ORM model.
- EmailPriorityEnum – Email priority enum for database.
- EmailStatusEnum – Email status enum for database.
# leadr.infra.email.adapters.orm.EmailORM¶
Bases: Base
Email ORM model.
Represents an email in the database for tracking and auditing purposes. Maps to the emails table.
Functions:
- from_domain – Convert domain entity to ORM model.
- to_domain – Convert ORM model to domain entity.
Attributes:
- bcc (
Mapped[list[str]]) – - body (
Mapped[str]) – - cc (
Mapped[list[str]]) – - created_at (
Mapped[timestamp]) – - deleted_at (
Mapped[nullable_timestamp]) – - error_message (
Mapped[str | None]) – - failed_at (
Mapped[datetime | None]) – - from_email (
Mapped[str | None]) – - id (
Mapped[uuid_pk]) – - priority (
Mapped[EmailPriorityEnum]) – - provider_message_id (
Mapped[str | None]) – - provider_response (
Mapped[dict[str, Any] | None]) – - reply_to (
Mapped[str | None]) – - sent_at (
Mapped[datetime | None]) – - status (
Mapped[EmailStatusEnum]) – - subject (
Mapped[str]) – - template_data (
Mapped[dict[str, Any] | None]) – - to (
Mapped[str]) – - updated_at (
Mapped[timestamp]) –
## leadr.infra.email.adapters.orm.EmailORM.bcc¶
bcc: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list, server_default='[]')
## leadr.infra.email.adapters.orm.EmailORM.body¶
body: Mapped[str] = mapped_column(Text, nullable=False)
## leadr.infra.email.adapters.orm.EmailORM.cc¶
cc: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list, server_default='[]')
## leadr.infra.email.adapters.orm.EmailORM.created_at¶
created_at: Mapped[timestamp]
## leadr.infra.email.adapters.orm.EmailORM.deleted_at¶
deleted_at: Mapped[nullable_timestamp]
## leadr.infra.email.adapters.orm.EmailORM.error_message¶
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.failed_at¶
failed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.from_domain¶
from_domain(domain)
Convert domain entity to ORM model.
Parameters:
- domain (
Email) – The domain entity to convert.
Returns:
EmailORM– The ORM model instance.
## leadr.infra.email.adapters.orm.EmailORM.from_email¶
from_email: Mapped[str | None] = mapped_column(String, nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.id¶
id: Mapped[uuid_pk]
## leadr.infra.email.adapters.orm.EmailORM.priority¶
priority: Mapped[EmailPriorityEnum] = mapped_column(Enum(EmailPriorityEnum, name='email_priority', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, default=(EmailPriorityEnum.NORMAL), server_default='normal')
## leadr.infra.email.adapters.orm.EmailORM.provider_message_id¶
provider_message_id: Mapped[str | None] = mapped_column(String, nullable=True, index=True)
## leadr.infra.email.adapters.orm.EmailORM.provider_response¶
provider_response: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.reply_to¶
reply_to: Mapped[str | None] = mapped_column(String, nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.sent_at¶
sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.status¶
status: Mapped[EmailStatusEnum] = mapped_column(Enum(EmailStatusEnum, name='email_status', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, default=(EmailStatusEnum.PENDING), server_default='pending', index=True)
## leadr.infra.email.adapters.orm.EmailORM.subject¶
subject: Mapped[str] = mapped_column(String, nullable=False)
## leadr.infra.email.adapters.orm.EmailORM.template_data¶
template_data: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
## leadr.infra.email.adapters.orm.EmailORM.to¶
to: Mapped[str] = mapped_column(String, nullable=False, index=True)
## leadr.infra.email.adapters.orm.EmailORM.to_domain¶
to_domain()
Convert ORM model to domain entity.
Returns:
Email– The domain entity instance.
## leadr.infra.email.adapters.orm.EmailORM.updated_at¶
updated_at: Mapped[timestamp] = mapped_column(onupdate=(func.now()))
# leadr.infra.email.adapters.orm.EmailPriorityEnum¶
Email priority enum for database.
Attributes:
## leadr.infra.email.adapters.orm.EmailPriorityEnum.HIGH¶
HIGH = 'high'
## leadr.infra.email.adapters.orm.EmailPriorityEnum.LOW¶
LOW = 'low'
## leadr.infra.email.adapters.orm.EmailPriorityEnum.NORMAL¶
NORMAL = 'normal'
## leadr.infra.email.adapters.orm.EmailPriorityEnum.URGENT¶
URGENT = 'urgent'
# leadr.infra.email.adapters.orm.EmailStatusEnum¶
Email status enum for database.
Attributes:
## leadr.infra.email.adapters.orm.EmailStatusEnum.DELIVERED¶
DELIVERED = 'delivered'
## leadr.infra.email.adapters.orm.EmailStatusEnum.FAILED¶
FAILED = 'failed'
## leadr.infra.email.adapters.orm.EmailStatusEnum.PENDING¶
PENDING = 'pending'
## leadr.infra.email.adapters.orm.EmailStatusEnum.SENT¶
SENT = 'sent'
leadr.infra.email.adapters.repositories¶
Email repository for database operations.
Classes:
- EmailRepository – Repository for Email entities.
# leadr.infra.email.adapters.repositories.EmailRepository¶
EmailRepository(db)
Bases: BaseRepository[Email, EmailORM]
Repository for Email entities.
Functions:
- create – Create a new entity in the database.
- delete – Soft delete an entity by setting its deleted_at timestamp.
- filter – Filter emails by criteria.
- get_by_id – Get an entity by its ID.
- update – Update an existing entity in the database.
Attributes:
- session –
Parameters:
- db (
AsyncSession) – Database session.
## leadr.infra.email.adapters.repositories.EmailRepository.create¶
create(entity)
Create a new entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity to create
Returns:
DomainEntityT– Created domain entity with refreshed data
## leadr.infra.email.adapters.repositories.EmailRepository.delete¶
delete(entity_id)
Soft delete an entity by setting its deleted_at timestamp.
Parameters:
- entity_id (
UUID4 | PrefixedID) – ID of entity to delete
Raises:
EntityNotFoundError– If entity is not found
## leadr.infra.email.adapters.repositories.EmailRepository.filter¶
filter(account_id=None, **kwargs)
Filter emails by criteria.
Parameters:
- account_id (
Any | None) – Not used for emails (top-level entity). - **kwargs (
Any) – Filter parameters (to, status, etc.)
Returns:
## leadr.infra.email.adapters.repositories.EmailRepository.get_by_id¶
get_by_id(entity_id, include_deleted=False)
Get an entity by its ID.
Parameters:
- entity_id (
UUID4 | PrefixedID) – Entity ID to retrieve - include_deleted (
bool) – If True, include soft-deleted entities. Defaults to False.
Returns:
DomainEntityT | None– Domain entity if found, None otherwise
## leadr.infra.email.adapters.repositories.EmailRepository.session¶
session = session
## leadr.infra.email.adapters.repositories.EmailRepository.update¶
update(entity)
Update an existing entity in the database.
Parameters:
- entity (
DomainEntityT) – Domain entity with updated data
Returns:
DomainEntityT– Updated domain entity with refreshed data
Raises:
EntityNotFoundError– If entity is not found
leadr.infra.email.adapters.smtp¶
SMTP email adapter implementation for testing.
Classes:
- SMTPEmailProvider – SMTP email provider implementation for development/testing.
# leadr.infra.email.adapters.smtp.SMTPEmailProvider¶
SMTPEmailProvider(host=None, port=None)
Bases: EmailProvider
SMTP email provider implementation for development/testing.
Functions:
- send – Send email via SMTP.
- validate_config – Validate SMTP configuration.
Attributes:
Parameters:
- host (
str | None) – SMTP server hostname (defaults to settings.SMTP_HOST or localhost) - port (
int | None) – SMTP server port (defaults to settings.SMTP_PORT or 1025)
## leadr.infra.email.adapters.smtp.SMTPEmailProvider.host¶
host = host or getattr(settings, 'SMTP_HOST', 'localhost')
## leadr.infra.email.adapters.smtp.SMTPEmailProvider.port¶
port = port or getattr(settings, 'SMTP_PORT', 1025)
## leadr.infra.email.adapters.smtp.SMTPEmailProvider.send¶
send(email)
Send email via SMTP.
## leadr.infra.email.adapters.smtp.SMTPEmailProvider.validate_config¶
validate_config()
Validate SMTP configuration.
leadr.infra.email.create_email_service¶
create_email_service(provider=None, db=None)
Create an email service with the specified or default provider.
Parameters:
- provider (
EmailProvider | None) – Email provider instance. If None, uses MailgunEmailProvider in production or SMTPEmailProvider in TEST environment. - db (
AsyncSession | None) – Optional database session for persisting email records.
Returns:
EmailService– EmailService instance ready for use.
Example
# Use default provider (Mailgun in prod, SMTP in test) email_service = create_email_service() # With database persistence email_service = create_email_service(db=session) # Use custom provider custom_provider = MyCustomEmailProvider() email_service = create_email_service(provider=custom_provider)leadr.infra.email.domain¶
Email domain models and interfaces.
Modules:
- exceptions – Email domain exceptions.
- interfaces – Email domain interfaces.
- models – Email domain models.
leadr.infra.email.domain.exceptions¶
Email domain exceptions.
Classes:
- EmailError – Base exception for email domain errors.
- EmailSendError – Raised when email sending fails.
- EmailValidationError – Raised when email validation fails.
# leadr.infra.email.domain.exceptions.EmailError¶
Bases: Exception
Base exception for email domain errors.
# leadr.infra.email.domain.exceptions.EmailSendError¶
EmailSendError(message, provider_response=None)
Bases: EmailError
Raised when email sending fails.
Attributes:
## leadr.infra.email.domain.exceptions.EmailSendError.provider_response¶
provider_response = provider_response
# leadr.infra.email.domain.exceptions.EmailValidationError¶
Bases: EmailError
Raised when email validation fails.
leadr.infra.email.domain.interfaces¶
Email domain interfaces.
Classes:
- EmailProvider – Interface for email service providers.
# leadr.infra.email.domain.interfaces.EmailProvider¶
Bases: ABC
Interface for email service providers.
Functions:
- send – Send an email and return provider response.
- validate_config – Validate provider configuration.
## leadr.infra.email.domain.interfaces.EmailProvider.send¶
send(email)
Send an email and return provider response.
## leadr.infra.email.domain.interfaces.EmailProvider.validate_config¶
validate_config()
Validate provider configuration.
leadr.infra.email.domain.models¶
Email domain models.
Classes:
- Email – Email domain entity.
- EmailPriority – Email priority enumeration.
- EmailStatus – Email status enumeration.
# leadr.infra.email.domain.models.Email¶
Bases: Entity
Email domain entity.
Functions:
- create – Create a new Email entity.
- mark_as_delivered – Mark email as delivered.
- mark_as_failed – Mark email as failed.
- mark_as_sent – Mark email as sent.
- restore – Restore a soft-deleted entity.
- soft_delete – Mark entity as soft-deleted.
- validate_bcc_emails – Validate BCC email addresses.
- validate_body – Validate email body.
- validate_cc_emails – Validate CC email addresses.
- validate_from_email – Validate sender email address.
- validate_reply_to – Validate reply-to email address.
- validate_subject – Validate email subject.
- validate_to_email – Validate recipient email address.
Attributes:
- bcc (
list[str]) – - body (
str) – - cc (
list[str]) – - created_at (
datetime) – - deleted_at (
datetime | None) – - error_message (
str | None) – - failed_at (
datetime | None) – - from_email (
str | None) – - id (
EmailID) – - is_deleted (
bool) – Check if entity is soft-deleted. - model_config –
- priority (
EmailPriority) – - provider_message_id (
str | None) – - provider_response (
dict[str, Any] | None) – - reply_to (
str | None) – - sent_at (
datetime | None) – - status (
EmailStatus) – - subject (
str) – - template_data (
dict[str, Any] | None) – - to (
str) – - updated_at (
datetime) –
## leadr.infra.email.domain.models.Email.bcc¶
bcc: list[str] = Field(default_factory=list, description='BCC recipients')
## leadr.infra.email.domain.models.Email.body¶
body: str = Field(..., description='Email body content')
## leadr.infra.email.domain.models.Email.cc¶
cc: list[str] = Field(default_factory=list, description='CC recipients')
## leadr.infra.email.domain.models.Email.create¶
create(to, subject, body, from_email=None, reply_to=None, cc=None, bcc=None, priority=EmailPriority.NORMAL, template_data=None)
Create a new Email entity.
## leadr.infra.email.domain.models.Email.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
## leadr.infra.email.domain.models.Email.deleted_at¶
deleted_at: datetime | None = Field(default=None, description='Timestamp when entity was soft-deleted (UTC), or null if active')
## leadr.infra.email.domain.models.Email.error_message¶
error_message: str | None = Field(None, description='Error message if failed')
## leadr.infra.email.domain.models.Email.failed_at¶
failed_at: datetime | None = Field(None, description='Time email failed')
## leadr.infra.email.domain.models.Email.from_email¶
from_email: str | None = Field(None, description='Sender email address')
## leadr.infra.email.domain.models.Email.id¶
id: EmailID = Field(frozen=True, default_factory=EmailID, description='Unique email identifier')
## leadr.infra.email.domain.models.Email.is_deleted¶
is_deleted: bool
Check if entity is soft-deleted.
Returns:
bool– True if the entity has a deleted_at timestamp, False otherwise.
## leadr.infra.email.domain.models.Email.mark_as_delivered¶
mark_as_delivered()
Mark email as delivered.
## leadr.infra.email.domain.models.Email.mark_as_failed¶
mark_as_failed(error_message, provider_response=None)
Mark email as failed.
## leadr.infra.email.domain.models.Email.mark_as_sent¶
mark_as_sent(provider_message_id, provider_response)
Mark email as sent.
## leadr.infra.email.domain.models.Email.model_config¶
model_config = ConfigDict(validate_assignment=True)
## leadr.infra.email.domain.models.Email.priority¶
priority: EmailPriority = Field(default=(EmailPriority.NORMAL), description='Email priority')
## leadr.infra.email.domain.models.Email.provider_message_id¶
provider_message_id: str | None = Field(None, description='Provider message ID')
## leadr.infra.email.domain.models.Email.provider_response¶
provider_response: dict[str, Any] | None = Field(None, description='Provider API response')
## leadr.infra.email.domain.models.Email.reply_to¶
reply_to: str | None = Field(None, description='Reply-to email address')
## leadr.infra.email.domain.models.Email.restore¶
restore()
Restore a soft-deleted entity.
Clears the deleted_at timestamp, making the entity active again.
Example
> > > account.soft_delete() > > > account.restore() > > > assert account.is_deleted is False## leadr.infra.email.domain.models.Email.sent_at¶
sent_at: datetime | None = Field(None, description='Time email was sent')
## leadr.infra.email.domain.models.Email.soft_delete¶
soft_delete()
Mark entity as soft-deleted.
Sets the deleted_at timestamp to the current UTC time. Entities that are already deleted are not affected (deleted_at remains at original deletion time).
Example
> > > account = Account(name="Test", slug="test") > > > account.soft_delete() > > > assert account.is_deleted is True## leadr.infra.email.domain.models.Email.status¶
status: EmailStatus = Field(default=(EmailStatus.PENDING), description='Email status')
## leadr.infra.email.domain.models.Email.subject¶
subject: str = Field(..., description='Email subject')
## leadr.infra.email.domain.models.Email.template_data¶
template_data: dict[str, Any] | None = Field(None, description='Template data for email rendering')
## leadr.infra.email.domain.models.Email.to¶
to: str = Field(..., description='Recipient email address')
## leadr.infra.email.domain.models.Email.updated_at¶
updated_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp of last update (UTC)')
## leadr.infra.email.domain.models.Email.validate_bcc_emails¶
validate_bcc_emails(v)
Validate BCC email addresses.
## leadr.infra.email.domain.models.Email.validate_body¶
validate_body(v)
Validate email body.
## leadr.infra.email.domain.models.Email.validate_cc_emails¶
validate_cc_emails(v)
Validate CC email addresses.
## leadr.infra.email.domain.models.Email.validate_from_email¶
validate_from_email(v)
Validate sender email address.
## leadr.infra.email.domain.models.Email.validate_reply_to¶
validate_reply_to(v)
Validate reply-to email address.
## leadr.infra.email.domain.models.Email.validate_subject¶
validate_subject(v)
Validate email subject.
## leadr.infra.email.domain.models.Email.validate_to_email¶
validate_to_email(v)
Validate recipient email address.
# leadr.infra.email.domain.models.EmailPriority¶
Email priority enumeration.
Attributes:
## leadr.infra.email.domain.models.EmailPriority.HIGH¶
HIGH = 'high'
## leadr.infra.email.domain.models.EmailPriority.LOW¶
LOW = 'low'
## leadr.infra.email.domain.models.EmailPriority.NORMAL¶
NORMAL = 'normal'
## leadr.infra.email.domain.models.EmailPriority.URGENT¶
URGENT = 'urgent'
# leadr.infra.email.domain.models.EmailStatus¶
Email status enumeration.
Attributes:
## leadr.infra.email.domain.models.EmailStatus.DELIVERED¶
DELIVERED = 'delivered'
## leadr.infra.email.domain.models.EmailStatus.FAILED¶
FAILED = 'failed'
## leadr.infra.email.domain.models.EmailStatus.PENDING¶
PENDING = 'pending'
## leadr.infra.email.domain.models.EmailStatus.SENT¶
SENT = 'sent'
leadr.infra.email.service¶
Email service with dependency injection and convenience methods.
Classes:
- EmailService – Email service with dependency injection for testing and flexibility.
Attributes:
- logger –
leadr.infra.email.service.EmailService¶
EmailService(provider, db=None, validate_on_init=False)
Email service with dependency injection for testing and flexibility.
Functions:
- get_default_from_email – Get default from email address.
- send_email – Send an email using the configured provider.
- send_notification_email – Send a notification email.
- send_verification_code – Send a verification code email for LEADR registration.
- send_welcome_email – Send a welcome email after successful LEADR registration.
- validate_provider_config – Validate the email provider configuration.
Attributes:
- db –
- provider –
- repository –
- templates_dir –
Parameters:
- provider (
EmailProvider) – Email provider implementation (e.g., Mailgun). - db (
AsyncSession | None) – Optional database session for persisting email records. - validate_on_init (
bool) – Whether to validate provider config on initialization.
# leadr.infra.email.service.EmailService.db¶
db = db
# leadr.infra.email.service.EmailService.get_default_from_email¶
get_default_from_email()
Get default from email address.
# leadr.infra.email.service.EmailService.provider¶
provider = provider
# leadr.infra.email.service.EmailService.repository¶
repository = EmailRepository(db) if db else None
# leadr.infra.email.service.EmailService.send_email¶
send_email(to, subject, body, from_email=None, reply_to=None, cc=None, bcc=None, priority=EmailPriority.NORMAL, template_data=None)
Send an email using the configured provider.
# leadr.infra.email.service.EmailService.send_notification_email¶
send_notification_email(to, subject, message, priority=EmailPriority.NORMAL, from_email=None)
Send a notification email.
# leadr.infra.email.service.EmailService.send_verification_code¶
send_verification_code(to, code)
Send a verification code email for LEADR registration.
# leadr.infra.email.service.EmailService.send_welcome_email¶
send_welcome_email(to, user_name, account_name, account_slug, from_email=None)
Send a welcome email after successful LEADR registration.
# leadr.infra.email.service.EmailService.templates_dir¶
templates_dir = Path(__file__).parent / 'templates'
# leadr.infra.email.service.EmailService.validate_provider_config¶
validate_provider_config()
Validate the email provider configuration.
leadr.infra.email.service.logger¶
logger = logging.getLogger(__name__)