Infra
leadr.infra¶
Modules:
- blob_storage –
- cache – Cache infrastructure module.
- email – Email infrastructure domain with factory functions for easy integration.
- webhooks – Webhook event tracking infrastructure.
leadr.infra.blob_storage¶
Modules:
leadr.infra.blob_storage.adapters¶
leadr.infra.blob_storage.domain¶
leadr.infra.blob_storage.services¶
leadr.infra.cache¶
Cache infrastructure module.
Modules:
Classes:
- CacheBackend – Protocol for cache backends.
- InMemoryCache – Thread-safe in-memory cache with TTL support.
Functions:
- get_cache – Get the cache backend singleton.
Attributes:
- CacheDep –
leadr.infra.cache.CacheBackend¶
Bases: Protocol
Protocol for cache backends.
Defines the interface for cache implementations. This abstraction allows swapping cache backends (e.g., in-memory to Redis) without changing the consuming code.
Functions:
- delete – Delete a key from the cache.
- get – Get a value from the cache.
- set – Set a value in the cache with a TTL.
leadr.infra.cache.CacheBackend.delete¶
delete(key)
Delete a key from the cache.
Parameters:
- key (
str) – The cache key to delete.
leadr.infra.cache.CacheBackend.get¶
get(key)
Get a value from the cache.
Parameters:
- key (
str) – The cache key to retrieve.
Returns:
Any | None– The cached value, or None if not found or expired.
leadr.infra.cache.CacheBackend.set¶
set(key, value, ttl_seconds)
Set a value in the cache with a TTL.
Parameters:
- key (
str) – The cache key. - value (
Any) – The value to cache. - ttl_seconds (
int) – Time-to-live in seconds.
leadr.infra.cache.CacheDep¶
CacheDep = Annotated[CacheBackend, Depends(get_cache)]
leadr.infra.cache.InMemoryCache¶
InMemoryCache()
Thread-safe in-memory cache with TTL support.
This implementation uses a simple dict with expiration checking on read. Suitable for single-process applications. For multi-process or distributed deployments, replace with a Redis-backed implementation.
Usage
# Create a new instance cache = InMemoryCache() cache.set("key", "value", ttl_seconds=60) value = cache.get("key") # Or use the singleton for application-wide caching cache = InMemoryCache.get_instance()Functions:
- clear – Clear all entries from the cache.
- delete – Delete a key from the cache.
- get – Get a value from the cache.
- get_instance – Get the singleton instance.
- reset_instance – Reset the singleton instance.
- set – Set a value in the cache with a TTL.
leadr.infra.cache.InMemoryCache.clear¶
clear()
Clear all entries from the cache.
leadr.infra.cache.InMemoryCache.delete¶
delete(key)
Delete a key from the cache.
Parameters:
- key (
str) – The cache key to delete.
leadr.infra.cache.InMemoryCache.get¶
get(key)
Get a value from the cache.
Parameters:
- key (
str) – The cache key to retrieve.
Returns:
Any | None– The cached value, or None if not found or expired.
leadr.infra.cache.InMemoryCache.get_instance¶
get_instance()
Get the singleton instance.
Thread-safe singleton pattern for application-wide caching.
Returns:
InMemoryCache– The singleton InMemoryCache instance.
leadr.infra.cache.InMemoryCache.reset_instance¶
reset_instance()
Reset the singleton instance.
Primarily useful for testing to ensure a clean state.
leadr.infra.cache.InMemoryCache.set¶
set(key, value, ttl_seconds)
Set a value in the cache with a TTL.
Parameters:
- key (
str) – The cache key. - value (
Any) – The value to cache. - ttl_seconds (
int) – Time-to-live in seconds.
leadr.infra.cache.adapters¶
Cache adapters module.
Modules:
- memory – In-memory cache implementation.
Classes:
- InMemoryCache – Thread-safe in-memory cache with TTL support.
leadr.infra.cache.adapters.InMemoryCache¶
InMemoryCache()
Thread-safe in-memory cache with TTL support.
This implementation uses a simple dict with expiration checking on read. Suitable for single-process applications. For multi-process or distributed deployments, replace with a Redis-backed implementation.
Usage
# Create a new instance cache = InMemoryCache() cache.set("key", "value", ttl_seconds=60) value = cache.get("key") # Or use the singleton for application-wide caching cache = InMemoryCache.get_instance()Functions:
- clear – Clear all entries from the cache.
- delete – Delete a key from the cache.
- get – Get a value from the cache.
- get_instance – Get the singleton instance.
- reset_instance – Reset the singleton instance.
- set – Set a value in the cache with a TTL.
# leadr.infra.cache.adapters.InMemoryCache.clear¶
clear()
Clear all entries from the cache.
# leadr.infra.cache.adapters.InMemoryCache.delete¶
delete(key)
Delete a key from the cache.
Parameters:
- key (
str) – The cache key to delete.
# leadr.infra.cache.adapters.InMemoryCache.get¶
get(key)
Get a value from the cache.
Parameters:
- key (
str) – The cache key to retrieve.
Returns:
Any | None– The cached value, or None if not found or expired.
# leadr.infra.cache.adapters.InMemoryCache.get_instance¶
get_instance()
Get the singleton instance.
Thread-safe singleton pattern for application-wide caching.
Returns:
InMemoryCache– The singleton InMemoryCache instance.
# leadr.infra.cache.adapters.InMemoryCache.reset_instance¶
reset_instance()
Reset the singleton instance.
Primarily useful for testing to ensure a clean state.
# leadr.infra.cache.adapters.InMemoryCache.set¶
set(key, value, ttl_seconds)
Set a value in the cache with a TTL.
Parameters:
- key (
str) – The cache key. - value (
Any) – The value to cache. - ttl_seconds (
int) – Time-to-live in seconds.
leadr.infra.cache.adapters.memory¶
In-memory cache implementation.
Classes:
- CacheEntry – A cache entry with value and expiration time.
- InMemoryCache – Thread-safe in-memory cache with TTL support.
# leadr.infra.cache.adapters.memory.CacheEntry¶
CacheEntry(value, expires_at)
A cache entry with value and expiration time.
Attributes:
- expires_at (
float) – - value (
Any) –
## leadr.infra.cache.adapters.memory.CacheEntry.expires_at¶
expires_at: float
## leadr.infra.cache.adapters.memory.CacheEntry.value¶
value: Any
# leadr.infra.cache.adapters.memory.InMemoryCache¶
InMemoryCache()
Thread-safe in-memory cache with TTL support.
This implementation uses a simple dict with expiration checking on read. Suitable for single-process applications. For multi-process or distributed deployments, replace with a Redis-backed implementation.
Usage
# Create a new instance cache = InMemoryCache() cache.set("key", "value", ttl_seconds=60) value = cache.get("key") # Or use the singleton for application-wide caching cache = InMemoryCache.get_instance()Functions:
- clear – Clear all entries from the cache.
- delete – Delete a key from the cache.
- get – Get a value from the cache.
- get_instance – Get the singleton instance.
- reset_instance – Reset the singleton instance.
- set – Set a value in the cache with a TTL.
## leadr.infra.cache.adapters.memory.InMemoryCache.clear¶
clear()
Clear all entries from the cache.
## leadr.infra.cache.adapters.memory.InMemoryCache.delete¶
delete(key)
Delete a key from the cache.
Parameters:
- key (
str) – The cache key to delete.
## leadr.infra.cache.adapters.memory.InMemoryCache.get¶
get(key)
Get a value from the cache.
Parameters:
- key (
str) – The cache key to retrieve.
Returns:
Any | None– The cached value, or None if not found or expired.
## leadr.infra.cache.adapters.memory.InMemoryCache.get_instance¶
get_instance()
Get the singleton instance.
Thread-safe singleton pattern for application-wide caching.
Returns:
InMemoryCache– The singleton InMemoryCache instance.
## leadr.infra.cache.adapters.memory.InMemoryCache.reset_instance¶
reset_instance()
Reset the singleton instance.
Primarily useful for testing to ensure a clean state.
## leadr.infra.cache.adapters.memory.InMemoryCache.set¶
set(key, value, ttl_seconds)
Set a value in the cache with a TTL.
Parameters:
- key (
str) – The cache key. - value (
Any) – The value to cache. - ttl_seconds (
int) – Time-to-live in seconds.
leadr.infra.cache.domain¶
Cache domain module.
Modules:
- interfaces – Cache backend protocol definition.
Classes:
- CacheBackend – Protocol for cache backends.
leadr.infra.cache.domain.CacheBackend¶
Bases: Protocol
Protocol for cache backends.
Defines the interface for cache implementations. This abstraction allows swapping cache backends (e.g., in-memory to Redis) without changing the consuming code.
Functions:
- delete – Delete a key from the cache.
- get – Get a value from the cache.
- set – Set a value in the cache with a TTL.
# leadr.infra.cache.domain.CacheBackend.delete¶
delete(key)
Delete a key from the cache.
Parameters:
- key (
str) – The cache key to delete.
# leadr.infra.cache.domain.CacheBackend.get¶
get(key)
Get a value from the cache.
Parameters:
- key (
str) – The cache key to retrieve.
Returns:
Any | None– The cached value, or None if not found or expired.
# leadr.infra.cache.domain.CacheBackend.set¶
set(key, value, ttl_seconds)
Set a value in the cache with a TTL.
Parameters:
- key (
str) – The cache key. - value (
Any) – The value to cache. - ttl_seconds (
int) – Time-to-live in seconds.
leadr.infra.cache.domain.interfaces¶
Cache backend protocol definition.
Classes:
- CacheBackend – Protocol for cache backends.
# leadr.infra.cache.domain.interfaces.CacheBackend¶
Bases: Protocol
Protocol for cache backends.
Defines the interface for cache implementations. This abstraction allows swapping cache backends (e.g., in-memory to Redis) without changing the consuming code.
Functions:
- delete – Delete a key from the cache.
- get – Get a value from the cache.
- set – Set a value in the cache with a TTL.
## leadr.infra.cache.domain.interfaces.CacheBackend.delete¶
delete(key)
Delete a key from the cache.
Parameters:
- key (
str) – The cache key to delete.
## leadr.infra.cache.domain.interfaces.CacheBackend.get¶
get(key)
Get a value from the cache.
Parameters:
- key (
str) – The cache key to retrieve.
Returns:
Any | None– The cached value, or None if not found or expired.
## leadr.infra.cache.domain.interfaces.CacheBackend.set¶
set(key, value, ttl_seconds)
Set a value in the cache with a TTL.
Parameters:
- key (
str) – The cache key. - value (
Any) – The value to cache. - ttl_seconds (
int) – Time-to-live in seconds.
leadr.infra.cache.get_cache¶
get_cache()
Get the cache backend singleton.
Returns:
CacheBackend– The application-wide cache backend instance.
leadr.infra.cache.services¶
Cache services module.
Modules:
- dependencies – FastAPI dependencies for cache infrastructure.
Functions:
- get_cache – Get the cache backend singleton.
Attributes:
- CacheDep –
leadr.infra.cache.services.CacheDep¶
CacheDep = Annotated[CacheBackend, Depends(get_cache)]
leadr.infra.cache.services.dependencies¶
FastAPI dependencies for cache infrastructure.
Functions:
- get_cache – Get the cache backend singleton.
Attributes:
- CacheDep –
# leadr.infra.cache.services.dependencies.CacheDep¶
CacheDep = Annotated[CacheBackend, Depends(get_cache)]
# leadr.infra.cache.services.dependencies.get_cache¶
get_cache()
Get the cache backend singleton.
Returns:
CacheBackend– The application-wide cache backend instance.
leadr.infra.cache.services.get_cache¶
get_cache()
Get the cache backend singleton.
Returns:
CacheBackend– The application-wide cache backend instance.
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_invite_email – Send an invite email when a user is invited to an account.
- 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_invite_email¶
send_invite_email(to, account_name, code)
Send an invite email when a user is invited to an account.
Parameters:
- to (
str) – Email address to send the invite to. - account_name (
str) – Name of the account the user is being invited to. - code (
str) – The 6-character verification code.
Returns:
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), api_url=(settings.MAILGUN_API_URL))
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), api_url=(settings.MAILGUN_API_URL))
## 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 with pagination.
- get_by_id – Get an entity by its ID.
- update – Update an existing entity in the database.
Attributes:
Parameters:
- db (
AsyncSession) – Database session.
## leadr.infra.email.adapters.repositories.EmailRepository.SORTABLE_FIELDS¶
SORTABLE_FIELDS = {'id', 'to', 'status', 'priority', 'created_at', 'sent_at', 'updated_at'}
## 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, *, pagination, **kwargs)
Filter emails by criteria with pagination.
Parameters:
- account_id (
Any | None) – Not used for emails (top-level entity). - pagination (
PaginationParams) – Pagination parameters (required). - **kwargs (
Any) – Filter parameters (to, status, etc.)
Returns:
PaginatedResult[Email]– Paginated result of matching Email entities.
Raises:
ValueError– If sort field is not in SORTABLE_FIELDS.CursorValidationError– If cursor is invalid or state doesn't match.
## 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_invite_email – Send an invite email when a user is invited to an account.
- 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_invite_email¶
send_invite_email(to, account_name, code)
Send an invite email when a user is invited to an account.
Parameters:
- to (
str) – Email address to send the invite to. - account_name (
str) – Name of the account the user is being invited to. - code (
str) – The 6-character verification code.
Returns:
# 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__)
leadr.infra.webhooks¶
Webhook event tracking infrastructure.
Modules:
leadr.infra.webhooks.adapters¶
Webhook adapters.
Modules:
- orm – Webhook event ORM model.
leadr.infra.webhooks.adapters.orm¶
Webhook event ORM model.
Classes:
- WebhookEventORM – Webhook event ORM model.
- WebhookProcessingStatusEnum – Webhook processing status enum for database.
- WebhookSourceEnum – Webhook source enum for database.
# leadr.infra.webhooks.adapters.orm.WebhookEventORM¶
Bases: ImmutableBase
Webhook event ORM model.
Tracks received webhook events for idempotency and audit purposes. Uses a unique constraint on (source, external_event_id) to prevent duplicate processing.
Functions:
- from_domain – Convert domain entity to ORM model.
- to_domain – Convert ORM model to domain entity.
Attributes:
- created_at (
Mapped[timestamp]) – - error (
Mapped[str | None]) – - event_type (
Mapped[str]) – - external_event_id (
Mapped[str]) – - id (
Mapped[uuid_pk]) – - metadata –
- processed_at (
Mapped[datetime | None]) – - processing_status (
Mapped[WebhookProcessingStatusEnum]) – - registry –
- source (
Mapped[WebhookSourceEnum]) –
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.created_at¶
created_at: Mapped[timestamp]
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.error¶
error: Mapped[str | None] = mapped_column(Text, nullable=True)
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.event_type¶
event_type: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.external_event_id¶
external_event_id: Mapped[str] = mapped_column(String(256), nullable=False)
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.from_domain¶
from_domain(entity)
Convert domain entity to ORM model.
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.id¶
id: Mapped[uuid_pk]
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.metadata¶
metadata = Base.metadata
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.processed_at¶
processed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.processing_status¶
processing_status: Mapped[WebhookProcessingStatusEnum] = mapped_column(Enum(WebhookProcessingStatusEnum, name='webhook_processing_status', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, default=(WebhookProcessingStatusEnum.PENDING), server_default='pending', index=True)
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.registry¶
registry = Base.registry
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.source¶
source: Mapped[WebhookSourceEnum] = mapped_column(Enum(WebhookSourceEnum, name='webhook_source', native_enum=True, values_callable=(lambda x: [(e.value) for e in x])), nullable=False, index=True)
## leadr.infra.webhooks.adapters.orm.WebhookEventORM.to_domain¶
to_domain()
Convert ORM model to domain entity.
# leadr.infra.webhooks.adapters.orm.WebhookProcessingStatusEnum¶
Webhook processing status enum for database.
Attributes:
## leadr.infra.webhooks.adapters.orm.WebhookProcessingStatusEnum.FAILED¶
FAILED = 'failed'
## leadr.infra.webhooks.adapters.orm.WebhookProcessingStatusEnum.PENDING¶
PENDING = 'pending'
## leadr.infra.webhooks.adapters.orm.WebhookProcessingStatusEnum.PROCESSED¶
PROCESSED = 'processed'
# leadr.infra.webhooks.adapters.orm.WebhookSourceEnum¶
Webhook source enum for database.
Attributes:
- STRIPE –
## leadr.infra.webhooks.adapters.orm.WebhookSourceEnum.STRIPE¶
STRIPE = 'stripe'
leadr.infra.webhooks.domain¶
Webhook domain models.
Modules:
- enums – Webhook domain enums.
- ids – Webhook domain IDs.
- webhook_event – WebhookEvent domain entity.
leadr.infra.webhooks.domain.enums¶
Webhook domain enums.
Classes:
- WebhookProcessingStatus – Processing status of a received webhook event.
- WebhookSource – Source system that sent the webhook.
# leadr.infra.webhooks.domain.enums.WebhookProcessingStatus¶
Processing status of a received webhook event.
Attributes:
## leadr.infra.webhooks.domain.enums.WebhookProcessingStatus.FAILED¶
FAILED = 'failed'
## leadr.infra.webhooks.domain.enums.WebhookProcessingStatus.PENDING¶
PENDING = 'pending'
## leadr.infra.webhooks.domain.enums.WebhookProcessingStatus.PROCESSED¶
PROCESSED = 'processed'
# leadr.infra.webhooks.domain.enums.WebhookSource¶
Source system that sent the webhook.
Attributes:
- STRIPE –
## leadr.infra.webhooks.domain.enums.WebhookSource.STRIPE¶
STRIPE = 'stripe'
leadr.infra.webhooks.domain.ids¶
Webhook domain IDs.
Classes:
- WebhookEventID – Webhook event entity identifier.
# leadr.infra.webhooks.domain.ids.WebhookEventID¶
Bases: PrefixedID
Webhook event entity identifier.
Attributes:
## leadr.infra.webhooks.domain.ids.WebhookEventID.prefix¶
prefix = 'whe'
## leadr.infra.webhooks.domain.ids.WebhookEventID.uuid¶
uuid = uuid4()
leadr.infra.webhooks.domain.webhook_event¶
WebhookEvent domain entity.
Classes:
- WebhookEvent – Represents a received webhook event from an external system.
# leadr.infra.webhooks.domain.webhook_event.WebhookEvent¶
Bases: ImmutableEntity
Represents a received webhook event from an external system.
Tracks idempotency and processing status for external webhook events. Immutable after creation — processing state is updated via repository methods.
Attributes:
- created_at (
datetime) – - error (
str | None) – - event_type (
str) – - external_event_id (
str) – - id (
WebhookEventID) – - model_config –
- processed_at (
datetime | None) – - processing_status (
WebhookProcessingStatus) – - source (
WebhookSource) –
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.created_at¶
created_at: datetime = Field(default_factory=(lambda: datetime.now(UTC)), description='Timestamp when entity was created (UTC)')
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.error¶
error: str | None = Field(default=None, description='Error message if processing failed')
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.event_type¶
event_type: str = Field(description="The event type (e.g. 'customer.subscription.created')")
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.external_event_id¶
external_event_id: str = Field(description='The event ID from the external system (e.g. Stripe event ID)')
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.id¶
id: WebhookEventID = Field(frozen=True, default_factory=WebhookEventID, description='Unique identifier for this webhook event record')
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.model_config¶
model_config = ConfigDict(validate_assignment=True)
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.processed_at¶
processed_at: datetime | None = Field(default=None, description='Timestamp when event was successfully processed (UTC)')
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.processing_status¶
processing_status: WebhookProcessingStatus = Field(default=(WebhookProcessingStatus.PENDING), description='Current processing status of this event')
## leadr.infra.webhooks.domain.webhook_event.WebhookEvent.source¶
source: WebhookSource = Field(description='The external system that sent this webhook')
leadr.infra.webhooks.services¶
Webhook services.
Modules:
- dependencies – Webhook event service dependency injection.
- repository – Webhook event repository.
- webhook_service – Webhook event service for idempotency tracking.
leadr.infra.webhooks.services.dependencies¶
Webhook event service dependency injection.
Functions:
- get_webhook_event_service – Get WebhookEventService dependency.
Attributes:
# leadr.infra.webhooks.services.dependencies.WebhookEventServiceDep¶
WebhookEventServiceDep = Annotated[WebhookEventService, Depends(get_webhook_event_service)]
# leadr.infra.webhooks.services.dependencies.get_webhook_event_service¶
get_webhook_event_service(db)
Get WebhookEventService dependency.
leadr.infra.webhooks.services.repository¶
Webhook event repository.
Classes:
- WebhookEventRepository – Repository for WebhookEvent entities.
# leadr.infra.webhooks.services.repository.WebhookEventRepository¶
WebhookEventRepository(session)
Repository for WebhookEvent entities.
Handles creation and status tracking of incoming webhook events. Uses direct SQL updates for status transitions (mark_processed, mark_failed).
Functions:
- create – Persist a new webhook event record.
- get_by_source_and_external_id – Look up a webhook event by (source, external_event_id) — the idempotency key.
- mark_failed – Update the event to FAILED status with an error message.
- mark_processed – Update the event to PROCESSED status.
## leadr.infra.webhooks.services.repository.WebhookEventRepository.create¶
create(event)
Persist a new webhook event record.
## leadr.infra.webhooks.services.repository.WebhookEventRepository.get_by_source_and_external_id¶
get_by_source_and_external_id(source, external_event_id)
Look up a webhook event by (source, external_event_id) — the idempotency key.
## leadr.infra.webhooks.services.repository.WebhookEventRepository.mark_failed¶
mark_failed(event_id, error)
Update the event to FAILED status with an error message.
## leadr.infra.webhooks.services.repository.WebhookEventRepository.mark_processed¶
mark_processed(event_id, processed_at=None)
Update the event to PROCESSED status.
leadr.infra.webhooks.services.webhook_service¶
Webhook event service for idempotency tracking.
Classes:
- WebhookEventService – Service for tracking webhook event receipt and processing status.
# leadr.infra.webhooks.services.webhook_service.WebhookEventService¶
WebhookEventService(session)
Service for tracking webhook event receipt and processing status.
Provides idempotency by recording each event before processing and marking it as processed or failed afterwards. Callers should check is_already_processed() before performing domain logic to avoid duplicate side-effects.
Functions:
- get_existing_event – Return an existing webhook event record if one exists (any status).
- is_already_processed – Return True if this event has already been successfully processed.
- mark_failed – Mark a webhook event as failed with an error message.
- mark_processed – Mark a webhook event as successfully processed.
- record_received – Record that a webhook event has been received (status: PENDING).
## leadr.infra.webhooks.services.webhook_service.WebhookEventService.get_existing_event¶
get_existing_event(source, external_event_id)
Return an existing webhook event record if one exists (any status).
## leadr.infra.webhooks.services.webhook_service.WebhookEventService.is_already_processed¶
is_already_processed(source, external_event_id)
Return True if this event has already been successfully processed.
A PENDING or FAILED event is not considered processed — it can be retried.
## leadr.infra.webhooks.services.webhook_service.WebhookEventService.mark_failed¶
mark_failed(event_id, error)
Mark a webhook event as failed with an error message.
## leadr.infra.webhooks.services.webhook_service.WebhookEventService.mark_processed¶
mark_processed(event_id, processed_at=None)
Mark a webhook event as successfully processed.
## leadr.infra.webhooks.services.webhook_service.WebhookEventService.record_received¶
record_received(source, external_event_id, event_type)
Record that a webhook event has been received (status: PENDING).