API Reference¶

This section contains the complete API reference.

Main Interface¶

SubscriptionKore - Provider-agnostic async subscriptionkore management library.

class subscriptionkore.AppliedDiscount(*, discount_id: str, coupon_code: str | None = None, amount_off: Money | None = None, percent_off: Decimal | None = None, valid_until: datetime | None = None)[source]

Bases: BaseModel

Discount applied to a subscriptionn.

amount_off: Money | None

coupon_code: str | None

discount_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount_off': FieldInfo(annotation=Union[Money, NoneType], required=False, default=None), 'coupon_code': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'discount_id': FieldInfo(annotation=str, required=True), 'percent_off': FieldInfo(annotation=Union[Decimal, NoneType], required=False, default=None), 'valid_until': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

percent_off: Decimal | None

valid_until: datetime | None

class subscriptionkore.BillingPeriod(*, interval: Interval, interval_count: Annotated[int, Ge(ge=1)] = 1, anchor_date: datetime | None = None)[source]

Bases: BaseModel

Billing period configuration.

anchor_date: datetime | None

property display_name: str

interval: Interval

interval_count: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'anchor_date': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'interval': FieldInfo(annotation=Interval, required=True), 'interval_count': FieldInfo(annotation=int, required=False, default=1, metadata=[Ge(ge=1)])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

classmethod monthly() → BillingPeriod[source]

classmethod weekly() → BillingPeriod[source]

classmethod yearly() → BillingPeriod[source]

exception subscriptionkore.ConfigurationError(message: str, details: dict[str, Any] | None = None)[source]

Bases: SubscriptionKoreError

Base class for configuration errors.

class subscriptionkore.Customer(*, id: str = None, external_id: str, email: EmailStr, name: str | None = None, provider_refs: list[ProviderReference] = None, tax_info: TaxInfo | None = None, billing_address: Address | None = None, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]

Bases: BaseModel

Customer domain entity.

add_provider_ref(ref: ProviderReference) → None[source]: Add or update a provider reference.

billing_address: Address | None

created_at: datetime

email: EmailStr

external_id: str

get_provider_ref(provider_type: str) → ProviderReference | None[source]: Get provider reference for a specific provider.

id: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'billing_address': FieldInfo(annotation=Union[Address, NoneType], required=False, default=None), 'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'email': FieldInfo(annotation=EmailStr, required=True), 'external_id': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'name': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'provider_refs': FieldInfo(annotation=list[ProviderReference], required=False, default_factory=list), 'tax_info': FieldInfo(annotation=Union[TaxInfo, NoneType], required=False, default=None), 'updated_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str | None

provider_refs: list[ProviderReference]

tax_info: TaxInfo | None

update(email: str | None = None, name: str | None = None, tax_info: TaxInfo | None = None, billing_address: Address | None = None, metadata: dict[str, Any] | None = None) → None[source]: Update customer fields.

updated_at: datetime

class subscriptionkore.CustomerCreated(*, event_id: str = None, event_type: str = 'CustomerCreated', occurred_at: datetime = None, metadata: dict[str, Any] = None, customer: Customer)[source]

Bases: DomainEvent

Emitted when a customer is created.

customer: Customer

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer': FieldInfo(annotation=Customer, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='CustomerCreated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class subscriptionkore.CustomerEntitlement(*, customer_id: str, entitlement_key: str, current_value: bool | int | str, value_type: EntitlementValueType, source: EntitlementSource, expires_at: datetime | None = None, subscription_id: str | None = None, plan_id: str | None = None)[source]

Bases: BaseModel

Resolved entitlement for a customer.

as_bool() → bool[source]: Get value as boolean.

as_int() → int | None[source]: Get value as integer (None for unlimited).

current_value: bool | int | str

customer_id: str

entitlement_key: str

expires_at: datetime | None

is_boolean() → bool[source]

is_expired() → bool[source]: Check if entitlement has expired.

is_numeric() → bool[source]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'current_value': FieldInfo(annotation=Union[bool, int, str], required=True), 'customer_id': FieldInfo(annotation=str, required=True), 'entitlement_key': FieldInfo(annotation=str, required=True), 'expires_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'plan_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'source': FieldInfo(annotation=EntitlementSource, required=True), 'subscription_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'value_type': FieldInfo(annotation=EntitlementValueType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

plan_id: str | None

source: EntitlementSource

subscription_id: str | None

value_type: EntitlementValueType

class subscriptionkore.CustomerUpdated(*, event_id: str = None, event_type: str = 'CustomerUpdated', occurred_at: datetime = None, metadata: dict[str, Any] = None, customer: Customer, changed_fields: list[str])[source]

Bases: DomainEvent

Emitted when a customer is updated.

changed_fields: list[str]

customer: Customer

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'changed_fields': FieldInfo(annotation=list[str], required=True), 'customer': FieldInfo(annotation=Customer, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='CustomerUpdated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

exception subscriptionkore.DomainError(message: str, details: dict[str, Any] | None = None)[source]

Bases: SubscriptionKoreError

Base class for domain errors.

class subscriptionkore.DomainEvent(*, event_id: str = None, event_type: str = '', occurred_at: datetime = None, metadata: dict[str, Any] = None)[source]

Bases: BaseModel

Base class for all domain events.

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPlanChanged'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

exception subscriptionkore.DuplicateEntityError(entity_type: str, identifier: str)[source]

Bases: DomainError

Raised when attempting to create a duplicate entity.

class subscriptionkore.Entitlement(*, id: str = None, key: str, name: str, description: str | None = None, value_type: EntitlementValueType, default_value: bool | int | str | None = None, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]

Bases: BaseModel

Entitlement definition entity.

created_at: datetime

default_value: bool | int | str | None

description: str | None

id: str

key: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'default_value': FieldInfo(annotation=Union[bool, int, str, NoneType], required=False, default=None), 'description': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'key': FieldInfo(annotation=str, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'name': FieldInfo(annotation=str, required=True), 'updated_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'value_type': FieldInfo(annotation=EntitlementValueType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str

updated_at: datetime

value_type: EntitlementValueType

exception subscriptionkore.EntitlementError(message: str, details: dict[str, Any] | None = None)[source]

Bases: SubscriptionKoreError

Base class for entitlement errors.

exception subscriptionkore.EntitlementNotFoundError(key: str)[source]

Bases: EntitlementError

Raised when an entitlement is not found.

class subscriptionkore.EntitlementOverride(*, id: str = None, customer_id: str, entitlement_key: str, value: bool | int | str, value_type: EntitlementValueType, reason: str | None = None, expires_at: datetime | None = None, created_by: str | None = None, created_at: datetime = None)[source]

Bases: BaseModel

Manual entitlement override for a customer.

created_at: datetime

created_by: str | None

customer_id: str

entitlement_key: str

expires_at: datetime | None

id: str

is_expired() → bool[source]: Check if override has expired.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'created_by': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'customer_id': FieldInfo(annotation=str, required=True), 'entitlement_key': FieldInfo(annotation=str, required=True), 'expires_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'reason': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'value': FieldInfo(annotation=Union[bool, int, str], required=True), 'value_type': FieldInfo(annotation=EntitlementValueType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

reason: str | None

value: bool | int | str

value_type: EntitlementValueType

class subscriptionkore.EntitlementValueType(*values)[source]

Bases: StrEnum

Types of entitlement values.

BOOLEAN = 'boolean'

NUMERIC = 'numeric'

STRING = 'string'

UNLIMITED = 'unlimited'

exception subscriptionkore.EntityNotFoundError(entity_type: str, entity_id: str)[source]

Bases: DomainError

Raised when an entity is not found.

class subscriptionkore.Interval(*values)[source]

Bases: StrEnum

Billing interval types.

DAY = 'day'

MONTH = 'month'

WEEK = 'week'

YEAR = 'year'

exception subscriptionkore.InvalidConfigurationError(message: str, details: dict[str, Any] | None = None)[source]

Bases: ConfigurationError

Raised when configuration is invalid.

exception subscriptionkore.InvalidStateTransitionError(from_state: SubscriptionStatus, to_state: SubscriptionStatus, reason: str | None = None)[source]

Bases: DomainError

Raised when an invalid state transition is attempted.

class subscriptionkore.Invoice(*, id: str = None, customer_id: str, subscriptionkore_id: str | None = None, provider_ref: ProviderReference, status: InvoiceStatus = InvoiceStatus.DRAFT, subtotal: Money, tax: Money = None, discount_amount: Money = None, total: Money, amount_paid: Money = None, amount_due: Money, currency: Currency = Currency.USD, line_items: list[InvoiceLineItem] = None, period: DateRange | None = None, due_date: datetime | None = None, paid_at: datetime | None = None, invoice_pdf_url: str | None = None, hosted_invoice_url: str | None = None, metadata: dict[str, Any] = None, created_at: datetime = None)[source]

Bases: BaseModel

Invoice domain entity.

amount_due: Money

amount_paid: Money

created_at: datetime

currency: Currency

customer_id: str

discount_amount: Money

due_date: datetime | None

hosted_invoice_url: str | None

id: str

invoice_pdf_url: str | None

is_open() → bool[source]: Check if invoice is open for payment.

is_overdue() → bool[source]: Check if invoice is past due date.

is_paid() → bool[source]: Check if invoice is fully paid.

line_items: list[InvoiceLineItem]

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount_due': FieldInfo(annotation=Money, required=True), 'amount_paid': FieldInfo(annotation=Money, required=False, default_factory=<lambda>), 'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'currency': FieldInfo(annotation=Currency, required=False, default=<Currency.USD: 'USD'>), 'customer_id': FieldInfo(annotation=str, required=True), 'discount_amount': FieldInfo(annotation=Money, required=False, default_factory=<lambda>), 'due_date': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'hosted_invoice_url': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'invoice_pdf_url': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'line_items': FieldInfo(annotation=list[InvoiceLineItem], required=False, default_factory=list), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'paid_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'period': FieldInfo(annotation=Union[DateRange, NoneType], required=False, default=None), 'provider_ref': FieldInfo(annotation=ProviderReference, required=True), 'status': FieldInfo(annotation=InvoiceStatus, required=False, default=<InvoiceStatus.DRAFT: 'draft'>), 'subscriptionkore_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'subtotal': FieldInfo(annotation=Money, required=True), 'tax': FieldInfo(annotation=Money, required=False, default_factory=<lambda>), 'total': FieldInfo(annotation=Money, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

paid_at: datetime | None

period: DateRange | None

provider_ref: ProviderReference

remaining_balance() → Money[source]: Get remaining balance to be paid.

status: InvoiceStatus

subscriptionkore_id: str | None

subtotal: Money

tax: Money

total: Money

class subscriptionkore.InvoiceCreated(*, event_id: str = None, event_type: str = 'InvoiceCreated', occurred_at: datetime = None, metadata: dict[str, Any] = None, invoice: Invoice, customer_id: str, subscription_id: str | None = None)[source]

Bases: DomainEvent

Emitted when an invoice is created.

customer_id: str

invoice: Invoice

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='InvoiceCreated'), 'invoice': FieldInfo(annotation=Invoice, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscription_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscription_id: str | None

class subscriptionkore.InvoiceLineItem(*, id: str = None, description: str, quantity: int = 1, unit_amount: Money, amount: Money, period: DateRange | None = None, proration: bool = False, metadata: dict[str, Any] = None)[source]

Bases: BaseModel

Line item on an invoice.

amount: Money

description: str

id: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount': FieldInfo(annotation=Money, required=True), 'description': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'period': FieldInfo(annotation=Union[DateRange, NoneType], required=False, default=None), 'proration': FieldInfo(annotation=bool, required=False, default=False), 'quantity': FieldInfo(annotation=int, required=False, default=1), 'unit_amount': FieldInfo(annotation=Money, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

period: DateRange | None

proration: bool

quantity: int

unit_amount: Money

class subscriptionkore.InvoicePaid(*, event_id: str = None, event_type: str = 'InvoicePaid', occurred_at: datetime = None, metadata: dict[str, Any] = None, invoice: Invoice, customer_id: str, subscription_id: str | None = None, amount_paid: Money)[source]

Bases: DomainEvent

Emitted when an invoice is paid.

amount_paid: Money

customer_id: str

invoice: Invoice

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount_paid': FieldInfo(annotation=Money, required=True), 'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='InvoicePaid'), 'invoice': FieldInfo(annotation=Invoice, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscription_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscription_id: str | None

class subscriptionkore.InvoiceStatus(*values)[source]

Bases: StrEnum

Invoice status values.

DRAFT = 'draft'

OPEN = 'open'

PAID = 'paid'

UNCOLLECTIBLE = 'uncollectible'

VOID = 'void'

exception subscriptionkore.MissingProviderCredentialsError(provider: str, missing_fields: list[str])[source]

Bases: ConfigurationError

Raised when provider credentials are missing.

class subscriptionkore.Money(*, amount: Decimal, currency: Currency = Currency.USD)[source]

Bases: BaseModel

Immutable monetary value with currency.

amount: Decimal

currency: Currency

classmethod from_cents(cents: int, currency: Currency = Currency.USD) → Money[source]: Create Money from minor units (cents).

is_negative() → bool[source]

is_positive() → bool[source]

is_zero() → bool[source]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount': FieldInfo(annotation=Decimal, required=True, description='Amount in major currency units'), 'currency': FieldInfo(annotation=Currency, required=False, default=<Currency.USD: 'USD'>)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

classmethod normalize_amount(v: Any) → Decimal[source]

to_cents() → int[source]: Convert to minor units (cents).

classmethod zero(currency: Currency = Currency.USD) → Money[source]

class subscriptionkore.PauseConfig(*, resumes_at: datetime | None = None, behavior: PauseBehavior = PauseBehavior.VOID)[source]

Bases: BaseModel

Subscription pause configuration.

behavior: PauseBehavior

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'behavior': FieldInfo(annotation=PauseBehavior, required=False, default=<PauseBehavior.VOID: 'void'>), 'resumes_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

resumes_at: datetime | None

class subscriptionkore.PaymentEvent(*, id: str = None, provider_ref: ProviderReference, customer_id: str, subscriptionkore_id: str | None = None, invoice_id: str | None = None, event_type: PaymentEventType, amount: Money, status: PaymentStatus, failure_reason: str | None = None, failure_code: str | None = None, payment_method_type: str | None = None, payment_method_last4: str | None = None, occurred_at: datetime = None, metadata: dict[str, Any] = None)[source]

Bases: BaseModel

Payment event domain entity.

amount: Money

customer_id: str

event_type: PaymentEventType

failure_code: str | None

failure_reason: str | None

id: str

invoice_id: str | None

is_failed() → bool[source]: Check if payment failed.

is_successful() → bool[source]: Check if payment succeeded.

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount': FieldInfo(annotation=Money, required=True), 'customer_id': FieldInfo(annotation=str, required=True), 'event_type': FieldInfo(annotation=PaymentEventType, required=True), 'failure_code': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'failure_reason': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'invoice_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'payment_method_last4': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'payment_method_type': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'provider_ref': FieldInfo(annotation=ProviderReference, required=True), 'status': FieldInfo(annotation=PaymentStatus, required=True), 'subscriptionkore_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

payment_method_last4: str | None

payment_method_type: str | None

provider_ref: ProviderReference

status: PaymentStatus

subscriptionkore_id: str | None

class subscriptionkore.PaymentEventType(*values)[source]

Bases: StrEnum

Payment event types.

PAYMENT_DISPUTED = 'payment_disputed'

PAYMENT_FAILED = 'payment_failed'

PAYMENT_REFUNDED = 'payment_refunded'

PAYMENT_SUCCEEDED = 'payment_succeeded'

class subscriptionkore.PaymentFailed(*, event_id: str = None, event_type: str = 'PaymentFailed', occurred_at: datetime = None, metadata: dict[str, Any] = None, payment_event: PaymentEvent, customer_id: str, subscription_id: str | None = None, invoice_id: str | None = None, amount: Money, failure_reason: str | None = None, failure_code: str | None = None, attempt_count: int = 1)[source]

Bases: DomainEvent

Emitted when a payment fails.

amount: Money

attempt_count: int

customer_id: str

failure_code: str | None

failure_reason: str | None

invoice_id: str | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'amount': FieldInfo(annotation=Money, required=True), 'attempt_count': FieldInfo(annotation=int, required=False, default=1), 'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='PaymentFailed'), 'failure_code': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'failure_reason': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'invoice_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'payment_event': FieldInfo(annotation=PaymentEvent, required=True), 'subscription_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

payment_event: PaymentEvent

subscription_id: str | None

class subscriptionkore.PaymentStatus(*values)[source]

Bases: StrEnum

Payment status values.

DISPUTED = 'disputed'

FAILED = 'failed'

PARTIALLY_REFUNDED = 'partially_refunded'

PENDING = 'pending'

REFUNDED = 'refunded'

SUCCEEDED = 'succeeded'

class subscriptionkore.PaymentSucceeded(*, event_id: str = None, event_type: str = 'PaymentSucceeded', occurred_at: datetime = None, metadata: dict[str, Any] = None, payment_event: PaymentEvent, customer_id: str, subscription_id: str | None = None, invoice_id: str | None = None, amount: Money)[source]

Bases: DomainEvent

Emitted when a payment succeeds.

amount: Money

customer_id: str

invoice_id: str | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

payment_event: PaymentEvent

subscription_id: str | None

class subscriptionkore.Plan(*, id: str = None, product_id: str, name: str, description: str | None = None, provider_refs: list[ProviderReference] = None, price: Money, billing_period: BillingPeriod, trial_period_days: int | None = None, entitlements: list[PlanEntitlement] = None, active: bool = True, tier: int = 0, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]

Bases: BaseModel

Plan domain entity (a pricing tier for a product).

active: bool

add_provider_ref(ref: ProviderReference) → None[source]: Add or update a provider reference.

billing_period: BillingPeriod

created_at: datetime

deactivate() → None[source]: Deactivate the plan.

description: str | None

entitlements: list[PlanEntitlement]

get_entitlement_value(key: str) → bool | int | str | None[source]: Get entitlement value by key.

get_provider_ref(provider_type: str) → ProviderReference | None[source]: Get provider reference for a specific provider.

id: str

is_downgrade_from(other: Plan) → bool[source]: Check if this plan is a downgrade from another.

is_upgrade_from(other: Plan) → bool[source]: Check if this plan is an upgrade from another.

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'active': FieldInfo(annotation=bool, required=False, default=True), 'billing_period': FieldInfo(annotation=BillingPeriod, required=True), 'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'description': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'entitlements': FieldInfo(annotation=list[PlanEntitlement], required=False, default_factory=list), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'name': FieldInfo(annotation=str, required=True), 'price': FieldInfo(annotation=Money, required=True), 'product_id': FieldInfo(annotation=str, required=True), 'provider_refs': FieldInfo(annotation=list[ProviderReference], required=False, default_factory=list), 'tier': FieldInfo(annotation=int, required=False, default=0, description='For upgrade/downgrade ordering'), 'trial_period_days': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'updated_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str

price: Money

product_id: str

provider_refs: list[ProviderReference]

tier: int

trial_period_days: int | None

updated_at: datetime

class subscriptionkore.PlanEntitlement(*, entitlement_id: str, entitlement_key: str, value: bool | int | str | None, value_type: EntitlementValueType)[source]

Bases: BaseModel

Entitlement configuration for a plan.

entitlement_id: str

entitlement_key: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'entitlement_id': FieldInfo(annotation=str, required=True), 'entitlement_key': FieldInfo(annotation=str, required=True), 'value': FieldInfo(annotation=Union[bool, int, str, NoneType], required=True), 'value_type': FieldInfo(annotation=EntitlementValueType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

value: bool | int | str | None

value_type: EntitlementValueType

class subscriptionkore.Product(*, id: str = None, name: str, description: str | None = None, provider_refs: list[ProviderReference] = None, active: bool = True, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]

Bases: BaseModel

Product domain entity.

activate() → None[source]: Activate the product.

active: bool

add_provider_ref(ref: ProviderReference) → None[source]: Add or update a provider reference.

created_at: datetime

deactivate() → None[source]: Deactivate the product.

description: str | None

get_provider_ref(provider_type: str) → ProviderReference | None[source]: Get provider reference for a specific provider.

id: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'active': FieldInfo(annotation=bool, required=False, default=True), 'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'description': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'name': FieldInfo(annotation=str, required=True), 'provider_refs': FieldInfo(annotation=list[ProviderReference], required=False, default_factory=list), 'updated_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str

provider_refs: list[ProviderReference]

updated_at: datetime

exception subscriptionkore.ProviderAPIError(message: str, provider: str, status_code: int, provider_message: str | None = None, provider_code: str | None = None)[source]

Bases: ProviderError

Raised for non-recoverable API errors from providers.

exception subscriptionkore.ProviderAuthenticationError(provider: str, message: str | None = None)[source]

Bases: ProviderError

Raised when authentication fails. Non-recoverable.

exception subscriptionkore.ProviderError(message: str, provider: str, details: dict[str, Any] | None = None)[source]

Bases: SubscriptionKoreError

Base class for provider errors.

exception subscriptionkore.ProviderNetworkError(provider: str, original_error: Exception)[source]

Bases: ProviderError

Raised for network errors. Recoverable with retry.

exception subscriptionkore.ProviderRateLimitError(provider: str, retry_after: int | None = None)[source]

Bases: ProviderError

Raised when provider rate limit is exceeded. Recoverable with retry.

class subscriptionkore.ProviderReference(*, provider: ProviderType, external_id: str, metadata: dict[str, Any] = None)[source]

Bases: BaseModel

Reference to an entity in a payment provider.

external_id: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'external_id': FieldInfo(annotation=str, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'provider': FieldInfo(annotation=ProviderType, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

provider: ProviderType

class subscriptionkore.ProviderType(*values)[source]

Bases: StrEnum

Supported payment providers.

CHARGEBEE = 'chargebee'

LEMONSQUEEZY = 'lemonsqueezy'

PADDLE = 'paddle'

STRIPE = 'stripe'

exception subscriptionkore.RepositoryError(operation: str, entity_type: str, original_error: Exception)[source]

Bases: SubscriptionKoreError

Raised when a repository operation fails.

class subscriptionkore.Subscription(*, id: str = None, customer_id: str, plan_id: str, provider_ref: ProviderReference, status: SubscriptionStatus = SubscriptionStatus.INCOMPLETE, current_period: DateRange, trial_end: datetime | None = None, cancel_at_period_end: bool = False, canceled_at: datetime | None = None, ended_at: datetime | None = None, pause_collection: PauseConfig | None = None, discount: AppliedDiscount | None = None, quantity: int = 1, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]

Bases: BaseModel

Subscription domain entity.

apply_discount(discount: AppliedDiscount) → None[source]: Apply a discount to the subscriptionn.

cancel_at_period_end: bool

canceled_at: datetime | None

created_at: datetime

current_period: DateRange

customer_id: str

days_until_trial_ends() → int | None[source]: Get days until trial ends, or None if not trialing.

discount: AppliedDiscount | None

ended_at: datetime | None

has_ended() → bool[source]: Check if subscriptionn has permanently ended.

id: str

is_active() → bool[source]: Check if subscriptionn is in an active state.

is_canceled() → bool[source]: Check if subscriptionn is canceled.

is_past_due() → bool[source]: Check if subscriptionn has payment issues.

is_paused() → bool[source]: Check if subscriptionn is paused.

is_trialing() → bool[source]: Check if subscriptionn is in trial.

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cancel_at_period_end': FieldInfo(annotation=bool, required=False, default=False), 'canceled_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'created_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'current_period': FieldInfo(annotation=DateRange, required=True), 'customer_id': FieldInfo(annotation=str, required=True), 'discount': FieldInfo(annotation=Union[AppliedDiscount, NoneType], required=False, default=None), 'ended_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'pause_collection': FieldInfo(annotation=Union[PauseConfig, NoneType], required=False, default=None), 'plan_id': FieldInfo(annotation=str, required=True), 'provider_ref': FieldInfo(annotation=ProviderReference, required=True), 'quantity': FieldInfo(annotation=int, required=False, default=1), 'status': FieldInfo(annotation=SubscriptionStatus, required=False, default=<SubscriptionStatus.INCOMPLETE: 'incomplete'>), 'trial_end': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'updated_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

pause_collection: PauseConfig | None

plan_id: str

provider_ref: ProviderReference

quantity: int

remove_discount() → None[source]: Remove discount from subscriptionn.

schedule_cancellation() → None[source]: Schedule cancellation at period end.

status: SubscriptionStatus

trial_end: datetime | None

unschedule_cancellation() → None[source]: Remove scheduled cancellation.

updated_at: datetime

will_cancel_at_period_end() → bool[source]: Check if subscriptionn is scheduled to cancel.

class subscriptionkore.SubscriptionActivated(*, event_id: str = None, event_type: str = 'SubscriptionActivated', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, plan_id: str)[source]

Bases: DomainEvent

Emitted when a subscriptionkore becomes active.

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionActivated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'plan_id': FieldInfo(annotation=str, required=True), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

plan_id: str

subscriptionkore: Subscription

class subscriptionkore.SubscriptionCanceled(*, event_id: str = None, event_type: str = 'SubscriptionCanceled', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, plan_id: str, immediate: bool, reason: str | None = None)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is canceled.

customer_id: str

immediate: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionCanceled'), 'immediate': FieldInfo(annotation=bool, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'plan_id': FieldInfo(annotation=str, required=True), 'reason': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

plan_id: str

reason: str | None

subscriptionkore: Subscription

class subscriptionkore.SubscriptionCreated(*, event_id: str = None, event_type: str = 'SubscriptionCreated', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, plan_id: str)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is created.

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionCreated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'plan_id': FieldInfo(annotation=str, required=True), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

plan_id: str

subscriptionkore: Subscription

class subscriptionkore.SubscriptionKore(config: SubscriptionKoreConfig)[source]

Bases: object

Main entry point for the SubscriptionKore library.

Provides access to all subscription management functionality.

Example:

from subscriptionkore import SubscriptionKore, SubscriptionKoreConfig
from subscriptionkore.config import StripeConfig

config = SubscriptionKoreConfig(
    database_url="postgresql+asyncpg://user:pass@localhost/db",
    stripe=StripeConfig(
        api_key="sk_test_...",
        webhook_secret="whsec_...",
    ),
)

async with SubscriptionKore(config) as subscriptionkore:
    # Create a customer
    customer = await subscriptionkore.customers.create(
        external_id="user_123",
        email="user@example.com",
    )

    # Create a subscription
    subscription = await subscriptionkore.subscriptions.create(
        customer_id=customer.id,
        plan_id="plan_abc",
    )

    # Check entitlements
    has_access = await subscriptionkore.entitlements.has_access(
        customer_id=customer.id,
        entitlement_key="premium_features",
    )

async check_entitlement(customer_id: str, entitlement_key: str) → CustomerEntitlement[source]: Convenience method to check an entitlement.

async close() → None[source]: Close all connections and clean up resources.

property configured_providers: list[ProviderType]: Get list of configured providers.

async create_customer(external_id: str, email: str, name: str | None = None, provider_type: ProviderType | None = None) → Customer[source]: Convenience method to create a customer.

async create_subscription(customer_id: str, plan_id: str, quantity: int = 1, trial_period_days: int | None = None, provider_type: ProviderType | None = None) → Subscription[source]: Convenience method to create a subscription.

property event_bus: EventBusPort: Access the event bus for subscribing to domain events.

async get_customer_manager(provider_type: ProviderType | None = None) → CustomerManager[source]: Get a customer manager instance.

async get_entitlement_service() → EntitlementService[source]: Get an entitlement service instance.

get_provider(provider_type: ProviderType) → PaymentProviderPort[source]: Get a specific provider adapter.

async get_subscription_manager(provider_type: ProviderType | None = None) → SubscriptionManager[source]: Get a subscription manager instance.

async get_webhook_processor() → WebhookProcessor[source]: Get a webhook processor instance.

async has_access(customer_id: str, entitlement_key: str) → bool[source]: Convenience method to check feature access.

async initialize() → None[source]: Initialize all components.

on_event(event_type: type['DomainEvent']) → Callable[[Callable[['DomainEvent'], Awaitable[None]]], Callable[['DomainEvent'], Awaitable[None]]][source]

Decorator to subscribe to domain events.

Example:

@subscriptionkore.on_event(SubscriptionActivated)
async def handle_activation(event: SubscriptionActivated):
    await send_welcome_email(event.customer_id)

class subscriptionkore.SubscriptionKoreConfig(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | None = None, _cli_ignore_unknown_args: bool | None = None, _secrets_dir: PathType | None = None, *, database_url: str, redis_url: str | None = None, stripe: StripeConfig | None = None, paddle: PaddleConfig | None = None, lemonsqueezy: LemonSqueezyConfig | None = None, chargebee: ChargebeeConfig | None = None, default_provider: ProviderType = ProviderType.STRIPE, webhook_processing: Literal['sync', 'async'] = 'sync', entitlement_cache_ttl: int = 300, processed_event_ttl_days: int = 7)[source]

Bases: BaseSettings

Main configuration for SubscriptionKore.

chargebee: ChargebeeConfig | None

database_url: str

default_provider: ProviderType

entitlement_cache_ttl: int

get_configured_providers() → list[ProviderType][source]: Get list of configured providers.

get_provider_config(provider: ProviderType) → StripeConfig | PaddleConfig | LemonSqueezyConfig | ChargebeeConfig | None[source]: Get configuration for a specific provider.

lemonsqueezy: LemonSqueezyConfig | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'SUBSCRIPTIONKORE_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'chargebee': FieldInfo(annotation=Union[ChargebeeConfig, NoneType], required=False, default=None), 'database_url': FieldInfo(annotation=str, required=True, description='PostgreSQL async connection URL'), 'default_provider': FieldInfo(annotation=ProviderType, required=False, default=<ProviderType.STRIPE: 'stripe'>, description='Default payment provider to use'), 'entitlement_cache_ttl': FieldInfo(annotation=int, required=False, default=300, description='Entitlement cache TTL in seconds'), 'lemonsqueezy': FieldInfo(annotation=Union[LemonSqueezyConfig, NoneType], required=False, default=None), 'paddle': FieldInfo(annotation=Union[PaddleConfig, NoneType], required=False, default=None), 'processed_event_ttl_days': FieldInfo(annotation=int, required=False, default=7, description='Days to keep processed event records'), 'redis_url': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Redis connection URL'), 'stripe': FieldInfo(annotation=Union[StripeConfig, NoneType], required=False, default=None), 'webhook_processing': FieldInfo(annotation=Literal['sync', 'async'], required=False, default='sync', description='Webhook processing mode')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

paddle: PaddleConfig | None

processed_event_ttl_days: int

redis_url: str | None

stripe: StripeConfig | None

classmethod validate_database_url(v: str) → str[source]: Validate database URL is async-compatible.

validate_default_provider_configured() → SubscriptionKoreConfig[source]: Ensure default provider has configuration.

webhook_processing: Literal['sync', 'async']

exception subscriptionkore.SubscriptionKoreError(message: str, details: dict[str, Any] | None = None)[source]

Bases: Exception

Base exception for all subscriptionkore errors.

class subscriptionkore.SubscriptionPastDue(*, event_id: str = None, event_type: str = 'SubscriptionPastDue', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, invoice_id: str | None = None)[source]

Bases: DomainEvent

Emitted when a subscriptionkore becomes past due.

customer_id: str

invoice_id: str | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPastDue'), 'invoice_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscriptionkore: Subscription

class subscriptionkore.SubscriptionPaused(*, event_id: str = None, event_type: str = 'SubscriptionPaused', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, resumes_at: datetime | None = None)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is paused.

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPaused'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'resumes_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

resumes_at: datetime | None

subscriptionkore: Subscription

class subscriptionkore.SubscriptionPlanChanged(*, event_id: str = None, event_type: str = 'SubscriptionPlanChanged', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, previous_plan_id: str, new_plan_id: str, is_upgrade: bool)[source]

Bases: DomainEvent

Emitted when subscriptionkore plan is changed.

customer_id: str

is_upgrade: bool

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPlanChanged'), 'is_upgrade': FieldInfo(annotation=bool, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'new_plan_id': FieldInfo(annotation=str, required=True), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'previous_plan_id': FieldInfo(annotation=str, required=True), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

new_plan_id: str

previous_plan_id: str

subscriptionkore: Subscription

class subscriptionkore.SubscriptionResumed(*, event_id: str = None, event_type: str = 'SubscriptionResumed', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is resumed.

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionResumed'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscriptionkore: Subscription

class subscriptionkore.SubscriptionStatus(*values)[source]

Bases: StrEnum

Subscription status values.

ACTIVE = 'active'

CANCELED = 'canceled'

EXPIRED = 'expired'

INCOMPLETE = 'incomplete'

INCOMPLETE_EXPIRED = 'incomplete_expired'

PAST_DUE = 'past_due'

PAUSED = 'paused'

TRIALING = 'trialing'

UNPAID = 'unpaid'

class subscriptionkore.SubscriptionTrialEnded(*, event_id: str = None, event_type: str = 'SubscriptionTrialEnded', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, converted: bool)[source]

Bases: DomainEvent

Emitted when a trial ends.

converted: bool

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'converted': FieldInfo(annotation=bool, required=True), 'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionTrialEnded'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscriptionkore: Subscription

class subscriptionkore.SubscriptionTrialStarted(*, event_id: str = None, event_type: str = 'SubscriptionTrialStarted', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, trial_end: datetime)[source]

Bases: DomainEvent

Emitted when a trial starts.

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionTrialStarted'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True), 'trial_end': FieldInfo(annotation=datetime, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscriptionkore: Subscription

trial_end: datetime

class subscriptionkore.SubscriptionUpdated(*, event_id: str = None, event_type: str = 'SubscriptionUpdated', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, changed_fields: list[str])[source]

Bases: DomainEvent

Emitted when a subscriptionkore is updated.

changed_fields: list[str]

customer_id: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'changed_fields': FieldInfo(annotation=list[str], required=True), 'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionUpdated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

subscriptionkore: Subscription

exception subscriptionkore.UsageLimitExceededError(entitlement_key: str, limit: int, current_usage: int, requested: int)[source]

Bases: EntitlementError

Raised when a usage limit is exceeded.

exception subscriptionkore.ValidationError(field: str, message: str)[source]

Bases: DomainError

Raised when validation fails.

exception subscriptionkore.WebhookError(message: str, details: dict[str, Any] | None = None)[source]

Bases: SubscriptionKoreError

Base class for webhook errors.

exception subscriptionkore.WebhookPayloadInvalidError(provider: str, reason: str)[source]

Bases: WebhookError

Raised when webhook payload cannot be parsed.

exception subscriptionkore.WebhookProcessingError(event_id: str, reason: str)[source]

Bases: WebhookError

Raised when webhook processing fails.

exception subscriptionkore.WebhookSignatureInvalidError(provider: str)[source]

Bases: WebhookError

Raised when webhook signature verification fails.

Configuration¶

Configuration management.

Bases: BaseSettings

Chargebee provider configuration.

api_key: str¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'CHARGEBEE_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'api_key': FieldInfo(annotation=str, required=True, description='Chargebee API key'), 'site': FieldInfo(annotation=str, required=True, description='Chargebee site name'), 'webhook_password': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Webhook basic auth password'), 'webhook_username': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Webhook basic auth username')}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

site: str¶

webhook_password: str | None¶

webhook_username: str | None¶

class subscriptionkore.config.LemonSqueezyConfig(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | None = None, _cli_ignore_unknown_args: bool | None = None, _secrets_dir: PathType | None = None, *, api_key: str, webhook_secret: str, store_id: str)[source]¶

Bases: BaseSettings

LemonSqueezy provider configuration.

api_key: str¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'LEMONSQUEEZY_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'api_key': FieldInfo(annotation=str, required=True, description='LemonSqueezy API key'), 'store_id': FieldInfo(annotation=str, required=True, description='LemonSqueezy store ID'), 'webhook_secret': FieldInfo(annotation=str, required=True, description='LemonSqueezy webhook signing secret')}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

store_id: str¶

webhook_secret: str¶

class subscriptionkore.config.PaddleConfig(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | None = None, _cli_ignore_unknown_args: bool | None = None, _secrets_dir: PathType | None = None, *, api_key: str, webhook_secret: str, environment: Literal['sandbox', 'production'] = 'sandbox', seller_id: str | None = None)[source]¶

Bases: BaseSettings

Paddle provider configuration.

api_key: str¶

environment: Literal['sandbox', 'production']¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'PADDLE_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'api_key': FieldInfo(annotation=str, required=True, description='Paddle API key'), 'environment': FieldInfo(annotation=Literal['sandbox', 'production'], required=False, default='sandbox'), 'seller_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Paddle seller/vendor ID'), 'webhook_secret': FieldInfo(annotation=str, required=True, description='Paddle webhook secret key')}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

seller_id: str | None¶

webhook_secret: str¶

class subscriptionkore.config.StripeConfig(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | None = None, _cli_ignore_unknown_args: bool | None = None, _secrets_dir: PathType | None = None, *, api_key: str, webhook_secret: str, api_version: str = '2024-06-20')[source]¶

Bases: BaseSettings

Stripe provider configuration.

api_key: str¶

api_version: str¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'STRIPE_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'api_key': FieldInfo(annotation=str, required=True, description='Stripe secret API key'), 'api_version': FieldInfo(annotation=str, required=False, default='2024-06-20', description='Stripe API version'), 'webhook_secret': FieldInfo(annotation=str, required=True, description='Stripe webhook signing secret')}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

webhook_secret: str¶

class subscriptionkore.config.SubscriptionKoreConfig(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | None = None, _cli_ignore_unknown_args: bool | None = None, _secrets_dir: PathType | None = None, *, database_url: str, redis_url: str | None = None, stripe: StripeConfig | None = None, paddle: PaddleConfig | None = None, lemonsqueezy: LemonSqueezyConfig | None = None, chargebee: ChargebeeConfig | None = None, default_provider: ProviderType = ProviderType.STRIPE, webhook_processing: Literal['sync', 'async'] = 'sync', entitlement_cache_ttl: int = 300, processed_event_ttl_days: int = 7)[source]¶

Bases: BaseSettings

Main configuration for SubscriptionKore.

chargebee: ChargebeeConfig | None¶

database_url: str¶

default_provider: ProviderType¶

entitlement_cache_ttl: int¶

get_configured_providers() → list[ProviderType][source]¶: Get list of configured providers.

get_provider_config(provider: ProviderType) → StripeConfig | PaddleConfig | LemonSqueezyConfig | ChargebeeConfig | None[source]¶: Get configuration for a specific provider.

lemonsqueezy: LemonSqueezyConfig | None¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_use_class_docs_for_groups': False, 'env_file': None, 'env_file_encoding': None, 'env_ignore_empty': False, 'env_nested_delimiter': '__', 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'SUBSCRIPTIONKORE_', 'extra': 'forbid', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

paddle: PaddleConfig | None¶

processed_event_ttl_days: int¶

redis_url: str | None¶

stripe: StripeConfig | None¶

classmethod validate_database_url(v: str) → str[source]¶: Validate database URL is async-compatible.

validate_default_provider_configured() → SubscriptionKoreConfig[source]¶: Ensure default provider has configuration.

webhook_processing: Literal['sync', 'async']¶

Core Models¶

Subscription domain model.

class subscriptionkore.core.models.subscription.AppliedDiscount(*, discount_id: str, coupon_code: str | None = None, amount_off: Money | None = None, percent_off: Decimal | None = None, valid_until: datetime | None = None)[source]¶

Bases: BaseModel

Discount applied to a subscriptionn.

amount_off: Money | None¶

coupon_code: str | None¶

discount_id: str¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

percent_off: Decimal | None¶

valid_until: datetime | None¶

class subscriptionkore.core.models.subscription.PauseBehavior(*values)[source]¶

Bases: StrEnum

Pause behavior options.

KEEP_AS_DRAFT = 'keep_as_draft'¶

MARK_UNCOLLECTIBLE = 'mark_uncollectible'¶

VOID = 'void'¶

class subscriptionkore.core.models.subscription.PauseConfig(*, resumes_at: datetime | None = None, behavior: PauseBehavior = PauseBehavior.VOID)[source]¶

Bases: BaseModel

Subscription pause configuration.

behavior: PauseBehavior¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

resumes_at: datetime | None¶

class subscriptionkore.core.models.subscription.Subscription(*, id: str = None, customer_id: str, plan_id: str, provider_ref: ProviderReference, status: SubscriptionStatus = SubscriptionStatus.INCOMPLETE, current_period: DateRange, trial_end: datetime | None = None, cancel_at_period_end: bool = False, canceled_at: datetime | None = None, ended_at: datetime | None = None, pause_collection: PauseConfig | None = None, discount: AppliedDiscount | None = None, quantity: int = 1, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]¶

Bases: BaseModel

Subscription domain entity.

apply_discount(discount: AppliedDiscount) → None[source]¶: Apply a discount to the subscriptionn.

cancel_at_period_end: bool¶

canceled_at: datetime | None¶

created_at: datetime¶

current_period: DateRange¶

customer_id: str¶

days_until_trial_ends() → int | None[source]¶: Get days until trial ends, or None if not trialing.

discount: AppliedDiscount | None¶

ended_at: datetime | None¶

has_ended() → bool[source]¶: Check if subscriptionn has permanently ended.

id: str¶

is_active() → bool[source]¶: Check if subscriptionn is in an active state.

is_canceled() → bool[source]¶: Check if subscriptionn is canceled.

is_past_due() → bool[source]¶: Check if subscriptionn has payment issues.

is_paused() → bool[source]¶: Check if subscriptionn is paused.

is_trialing() → bool[source]¶: Check if subscriptionn is in trial.

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

pause_collection: PauseConfig | None¶

plan_id: str¶

provider_ref: ProviderReference¶

quantity: int¶

remove_discount() → None[source]¶: Remove discount from subscriptionn.

schedule_cancellation() → None[source]¶: Schedule cancellation at period end.

status: SubscriptionStatus¶

trial_end: datetime | None¶

unschedule_cancellation() → None[source]¶: Remove scheduled cancellation.

updated_at: datetime¶

will_cancel_at_period_end() → bool[source]¶: Check if subscriptionn is scheduled to cancel.

class subscriptionkore.core.models.subscription.SubscriptionStatus(*values)[source]¶

Bases: StrEnum

Subscription status values.

ACTIVE = 'active'¶

CANCELED = 'canceled'¶

EXPIRED = 'expired'¶

INCOMPLETE = 'incomplete'¶

INCOMPLETE_EXPIRED = 'incomplete_expired'¶

PAST_DUE = 'past_due'¶

PAUSED = 'paused'¶

TRIALING = 'trialing'¶

UNPAID = 'unpaid'¶

Customer domain model.

Bases: BaseModel

Billing address.

city: str | None¶

country: str | None¶

line1: str | None¶

line2: str | None¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'city': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'country': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'line1': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'line2': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'postal_code': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'state': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

postal_code: str | None¶

state: str | None¶

class subscriptionkore.core.models.customer.Customer(*, id: str = None, external_id: str, email: EmailStr, name: str | None = None, provider_refs: list[ProviderReference] = None, tax_info: TaxInfo | None = None, billing_address: Address | None = None, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]¶

Bases: BaseModel

Customer domain entity.

add_provider_ref(ref: ProviderReference) → None[source]¶: Add or update a provider reference.

billing_address: Address | None¶

created_at: datetime¶

email: EmailStr¶

external_id: str¶

get_provider_ref(provider_type: str) → ProviderReference | None[source]¶: Get provider reference for a specific provider.

id: str¶

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str | None¶

provider_refs: list[ProviderReference]¶

tax_info: TaxInfo | None¶

update(email: str | None = None, name: str | None = None, tax_info: TaxInfo | None = None, billing_address: Address | None = None, metadata: dict[str, Any] | None = None) → None[source]¶: Update customer fields.

updated_at: datetime¶

class subscriptionkore.core.models.customer.TaxInfo(*, tax_id: str | None = None, tax_id_type: str | None = None, tax_exempt: bool = False)[source]¶

Bases: BaseModel

Customer tax information.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'tax_exempt': FieldInfo(annotation=bool, required=False, default=False), 'tax_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'tax_id_type': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

tax_exempt: bool¶

tax_id: str | None¶

tax_id_type: str | None¶

Plan domain model.

class subscriptionkore.core.models.plan.Plan(*, id: str = None, product_id: str, name: str, description: str | None = None, provider_refs: list[ProviderReference] = None, price: Money, billing_period: BillingPeriod, trial_period_days: int | None = None, entitlements: list[PlanEntitlement] = None, active: bool = True, tier: int = 0, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]¶

Bases: BaseModel

Plan domain entity (a pricing tier for a product).

active: bool¶

add_provider_ref(ref: ProviderReference) → None[source]¶: Add or update a provider reference.

billing_period: BillingPeriod¶

created_at: datetime¶

deactivate() → None[source]¶: Deactivate the plan.

description: str | None¶

entitlements: list[PlanEntitlement]¶

get_entitlement_value(key: str) → bool | int | str | None[source]¶: Get entitlement value by key.

get_provider_ref(provider_type: str) → ProviderReference | None[source]¶: Get provider reference for a specific provider.

id: str¶

is_downgrade_from(other: Plan) → bool[source]¶: Check if this plan is a downgrade from another.

is_upgrade_from(other: Plan) → bool[source]¶: Check if this plan is an upgrade from another.

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str¶

price: Money¶

product_id: str¶

provider_refs: list[ProviderReference]¶

tier: int¶

trial_period_days: int | None¶

updated_at: datetime¶

class subscriptionkore.core.models.plan.PlanEntitlement(*, entitlement_id: str, entitlement_key: str, value: bool | int | str | None, value_type: EntitlementValueType)[source]¶

Bases: BaseModel

Entitlement configuration for a plan.

entitlement_id: str¶

entitlement_key: str¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

value: bool | int | str | None¶

value_type: EntitlementValueType¶

Product domain model.

class subscriptionkore.core.models.product.Product(*, id: str = None, name: str, description: str | None = None, provider_refs: list[ProviderReference] = None, active: bool = True, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]¶

Bases: BaseModel

Product domain entity.

activate() → None[source]¶: Activate the product.

active: bool¶

add_provider_ref(ref: ProviderReference) → None[source]¶: Add or update a provider reference.

created_at: datetime¶

deactivate() → None[source]¶: Deactivate the product.

description: str | None¶

get_provider_ref(provider_type: str) → ProviderReference | None[source]¶: Get provider reference for a specific provider.

id: str¶

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str¶

provider_refs: list[ProviderReference]¶

updated_at: datetime¶

Invoice domain model.

class subscriptionkore.core.models.invoice.Invoice(*, id: str = None, customer_id: str, subscriptionkore_id: str | None = None, provider_ref: ProviderReference, status: InvoiceStatus = InvoiceStatus.DRAFT, subtotal: Money, tax: Money = None, discount_amount: Money = None, total: Money, amount_paid: Money = None, amount_due: Money, currency: Currency = Currency.USD, line_items: list[InvoiceLineItem] = None, period: DateRange | None = None, due_date: datetime | None = None, paid_at: datetime | None = None, invoice_pdf_url: str | None = None, hosted_invoice_url: str | None = None, metadata: dict[str, Any] = None, created_at: datetime = None)[source]¶

Bases: BaseModel

Invoice domain entity.

amount_due: Money¶

amount_paid: Money¶

created_at: datetime¶

currency: Currency¶

customer_id: str¶

discount_amount: Money¶

due_date: datetime | None¶

hosted_invoice_url: str | None¶

id: str¶

invoice_pdf_url: str | None¶

is_open() → bool[source]¶: Check if invoice is open for payment.

is_overdue() → bool[source]¶: Check if invoice is past due date.

is_paid() → bool[source]¶: Check if invoice is fully paid.

line_items: list[InvoiceLineItem]¶

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

paid_at: datetime | None¶

period: DateRange | None¶

provider_ref: ProviderReference¶

remaining_balance() → Money[source]¶: Get remaining balance to be paid.

status: InvoiceStatus¶

subscriptionkore_id: str | None¶

subtotal: Money¶

tax: Money¶

total: Money¶

class subscriptionkore.core.models.invoice.InvoiceLineItem(*, id: str = None, description: str, quantity: int = 1, unit_amount: Money, amount: Money, period: DateRange | None = None, proration: bool = False, metadata: dict[str, Any] = None)[source]¶

Bases: BaseModel

Line item on an invoice.

amount: Money¶

description: str¶

id: str¶

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

period: DateRange | None¶

proration: bool¶

quantity: int¶

unit_amount: Money¶

class subscriptionkore.core.models.invoice.InvoiceStatus(*values)[source]¶

Bases: StrEnum

Invoice status values.

DRAFT = 'draft'¶

OPEN = 'open'¶

PAID = 'paid'¶

UNCOLLECTIBLE = 'uncollectible'¶

VOID = 'void'¶

Payment event domain model.

class subscriptionkore.core.models.payment_event.PaymentEvent(*, id: str = None, provider_ref: ProviderReference, customer_id: str, subscriptionkore_id: str | None = None, invoice_id: str | None = None, event_type: PaymentEventType, amount: Money, status: PaymentStatus, failure_reason: str | None = None, failure_code: str | None = None, payment_method_type: str | None = None, payment_method_last4: str | None = None, occurred_at: datetime = None, metadata: dict[str, Any] = None)[source]¶

Bases: BaseModel

Payment event domain entity.

amount: Money¶

customer_id: str¶

event_type: PaymentEventType¶

failure_code: str | None¶

failure_reason: str | None¶

id: str¶

invoice_id: str | None¶

is_failed() → bool[source]¶: Check if payment failed.

is_successful() → bool[source]¶: Check if payment succeeded.

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime¶

payment_method_last4: str | None¶

payment_method_type: str | None¶

provider_ref: ProviderReference¶

status: PaymentStatus¶

subscriptionkore_id: str | None¶

class subscriptionkore.core.models.payment_event.PaymentEventType(*values)[source]¶

Bases: StrEnum

Payment event types.

PAYMENT_DISPUTED = 'payment_disputed'¶

PAYMENT_FAILED = 'payment_failed'¶

PAYMENT_REFUNDED = 'payment_refunded'¶

PAYMENT_SUCCEEDED = 'payment_succeeded'¶

class subscriptionkore.core.models.payment_event.PaymentStatus(*values)[source]¶

Bases: StrEnum

Payment status values.

DISPUTED = 'disputed'¶

FAILED = 'failed'¶

PARTIALLY_REFUNDED = 'partially_refunded'¶

PENDING = 'pending'¶

REFUNDED = 'refunded'¶

SUCCEEDED = 'succeeded'¶

Entitlement domain models.

class subscriptionkore.core.models.entitlement.CustomerEntitlement(*, customer_id: str, entitlement_key: str, current_value: bool | int | str, value_type: EntitlementValueType, source: EntitlementSource, expires_at: datetime | None = None, subscription_id: str | None = None, plan_id: str | None = None)[source]¶

Bases: BaseModel

Resolved entitlement for a customer.

as_bool() → bool[source]¶: Get value as boolean.

as_int() → int | None[source]¶: Get value as integer (None for unlimited).

current_value: bool | int | str¶

customer_id: str¶

entitlement_key: str¶

expires_at: datetime | None¶

is_boolean() → bool[source]¶

is_expired() → bool[source]¶: Check if entitlement has expired.

is_numeric() → bool[source]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

plan_id: str | None¶

source: EntitlementSource¶

subscription_id: str | None¶

value_type: EntitlementValueType¶

class subscriptionkore.core.models.entitlement.Entitlement(*, id: str = None, key: str, name: str, description: str | None = None, value_type: EntitlementValueType, default_value: bool | int | str | None = None, metadata: dict[str, Any] = None, created_at: datetime = None, updated_at: datetime = None)[source]¶

Bases: BaseModel

Entitlement definition entity.

created_at: datetime¶

default_value: bool | int | str | None¶

description: str | None¶

id: str¶

key: str¶

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

name: str¶

updated_at: datetime¶

value_type: EntitlementValueType¶

class subscriptionkore.core.models.entitlement.EntitlementOverride(*, id: str = None, customer_id: str, entitlement_key: str, value: bool | int | str, value_type: EntitlementValueType, reason: str | None = None, expires_at: datetime | None = None, created_by: str | None = None, created_at: datetime = None)[source]¶

Bases: BaseModel

Manual entitlement override for a customer.

created_at: datetime¶

created_by: str | None¶

customer_id: str¶

entitlement_key: str¶

expires_at: datetime | None¶

id: str¶

is_expired() → bool[source]¶: Check if override has expired.

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

reason: str | None¶

value: bool | int | str¶

value_type: EntitlementValueType¶

class subscriptionkore.core.models.entitlement.EntitlementSource(*values)[source]¶

Bases: StrEnum

Source of an entitlement.

DEFAULT = 'default'¶

OVERRIDE = 'override'¶

PLAN = 'plan'¶

TRIAL = 'trial'¶

class subscriptionkore.core.models.entitlement.EntitlementValueType(*values)[source]¶

Bases: StrEnum

Types of entitlement values.

BOOLEAN = 'boolean'¶

NUMERIC = 'numeric'¶

STRING = 'string'¶

UNLIMITED = 'unlimited'¶

Value Objects¶

Value objects for the domain layer.

class subscriptionkore.core.models.value_objects.BillingPeriod(*, interval: Interval, interval_count: Annotated[int, Ge(ge=1)] = 1, anchor_date: datetime | None = None)[source]¶

Bases: BaseModel

Billing period configuration.

anchor_date: datetime | None¶

property display_name: str¶

interval: Interval¶

interval_count: int¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

classmethod monthly() → BillingPeriod[source]¶

classmethod weekly() → BillingPeriod[source]¶

classmethod yearly() → BillingPeriod[source]¶

class subscriptionkore.core.models.value_objects.Currency(*values)[source]¶

Bases: StrEnum

ISO 4217 currency codes.

AUD = 'AUD'¶

BRL = 'BRL'¶

CAD = 'CAD'¶

CHF = 'CHF'¶

EUR = 'EUR'¶

GBP = 'GBP'¶

INR = 'INR'¶

JPY = 'JPY'¶

MXN = 'MXN'¶

USD = 'USD'¶

class subscriptionkore.core.models.value_objects.DateRange(*, start: datetime, end: datetime | None = None)[source]¶

Bases: BaseModel

Date range with start and optional end.

contains(dt: datetime) → bool[source]¶

end: datetime | None¶

is_open_ended() → bool[source]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'end': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'start': FieldInfo(annotation=datetime, required=True)}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

start: datetime¶

validate_range() → Self[source]¶

class subscriptionkore.core.models.value_objects.Interval(*values)[source]¶

Bases: StrEnum

Billing interval types.

DAY = 'day'¶

MONTH = 'month'¶

WEEK = 'week'¶

YEAR = 'year'¶

class subscriptionkore.core.models.value_objects.Money(*, amount: Decimal, currency: Currency = Currency.USD)[source]¶

Bases: BaseModel

Immutable monetary value with currency.

amount: Decimal¶

currency: Currency¶

classmethod from_cents(cents: int, currency: Currency = Currency.USD) → Money[source]¶: Create Money from minor units (cents).

is_negative() → bool[source]¶

is_positive() → bool[source]¶

is_zero() → bool[source]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

classmethod normalize_amount(v: Any) → Decimal[source]¶

to_cents() → int[source]¶: Convert to minor units (cents).

classmethod zero(currency: Currency = Currency.USD) → Money[source]¶

class subscriptionkore.core.models.value_objects.ProviderReference(*, provider: ProviderType, external_id: str, metadata: dict[str, Any] = None)[source]¶

Bases: BaseModel

Reference to an entity in a payment provider.

external_id: str¶

metadata: dict[str, Any]¶

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

provider: ProviderType¶

class subscriptionkore.core.models.value_objects.ProviderType(*values)[source]¶

Bases: StrEnum

Supported payment providers.

CHARGEBEE = 'chargebee'¶

LEMONSQUEEZY = 'lemonsqueezy'¶

PADDLE = 'paddle'¶

STRIPE = 'stripe'¶

Events¶

Domain events.

class subscriptionkore.core.events.CustomerCreated(*, event_id: str = None, event_type: str = 'CustomerCreated', occurred_at: datetime = None, metadata: dict[str, Any] = None, customer: Customer)[source]

Bases: DomainEvent

Emitted when a customer is created.

customer: Customer

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

class subscriptionkore.core.events.CustomerUpdated(*, event_id: str = None, event_type: str = 'CustomerUpdated', occurred_at: datetime = None, metadata: dict[str, Any] = None, customer: Customer, changed_fields: list[str])[source]

Bases: DomainEvent

Emitted when a customer is updated.

changed_fields: list[str]

customer: Customer

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

class subscriptionkore.core.events.DomainEvent(*, event_id: str = None, event_type: str = '', occurred_at: datetime = None, metadata: dict[str, Any] = None)[source]

Bases: BaseModel

Base class for all domain events.

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

class subscriptionkore.core.events.InvoiceCreated(*, event_id: str = None, event_type: str = 'InvoiceCreated', occurred_at: datetime = None, metadata: dict[str, Any] = None, invoice: Invoice, customer_id: str, subscription_id: str | None = None)[source]

Bases: DomainEvent

Emitted when an invoice is created.

customer_id: str

event_id: str

event_type: str

invoice: Invoice

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscription_id: str | None

class subscriptionkore.core.events.InvoicePaid(*, event_id: str = None, event_type: str = 'InvoicePaid', occurred_at: datetime = None, metadata: dict[str, Any] = None, invoice: Invoice, customer_id: str, subscription_id: str | None = None, amount_paid: Money)[source]

Bases: DomainEvent

Emitted when an invoice is paid.

amount_paid: Money

customer_id: str

event_id: str

event_type: str

invoice: Invoice

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscription_id: str | None

class subscriptionkore.core.events.PaymentFailed(*, event_id: str = None, event_type: str = 'PaymentFailed', occurred_at: datetime = None, metadata: dict[str, Any] = None, payment_event: PaymentEvent, customer_id: str, subscription_id: str | None = None, invoice_id: str | None = None, amount: Money, failure_reason: str | None = None, failure_code: str | None = None, attempt_count: int = 1)[source]

Bases: DomainEvent

Emitted when a payment fails.

amount: Money

attempt_count: int

customer_id: str

event_id: str

event_type: str

failure_code: str | None

failure_reason: str | None

invoice_id: str | None

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

payment_event: PaymentEvent

subscription_id: str | None

class subscriptionkore.core.events.PaymentSucceeded(*, event_id: str = None, event_type: str = 'PaymentSucceeded', occurred_at: datetime = None, metadata: dict[str, Any] = None, payment_event: PaymentEvent, customer_id: str, subscription_id: str | None = None, invoice_id: str | None = None, amount: Money)[source]

Bases: DomainEvent

Emitted when a payment succeeds.

amount: Money

customer_id: str

event_id: str

event_type: str

invoice_id: str | None

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

payment_event: PaymentEvent

subscription_id: str | None

class subscriptionkore.core.events.SubscriptionActivated(*, event_id: str = None, event_type: str = 'SubscriptionActivated', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, plan_id: str)[source]

Bases: DomainEvent

Emitted when a subscriptionkore becomes active.

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionActivated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'plan_id': FieldInfo(annotation=str, required=True), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

plan_id: str

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionCanceled(*, event_id: str = None, event_type: str = 'SubscriptionCanceled', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, plan_id: str, immediate: bool, reason: str | None = None)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is canceled.

customer_id: str

event_id: str

event_type: str

immediate: bool

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionCanceled'), 'immediate': FieldInfo(annotation=bool, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'plan_id': FieldInfo(annotation=str, required=True), 'reason': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

plan_id: str

reason: str | None

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionCreated(*, event_id: str = None, event_type: str = 'SubscriptionCreated', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, plan_id: str)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is created.

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionCreated'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'plan_id': FieldInfo(annotation=str, required=True), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

plan_id: str

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionPastDue(*, event_id: str = None, event_type: str = 'SubscriptionPastDue', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, invoice_id: str | None = None)[source]

Bases: DomainEvent

Emitted when a subscriptionkore becomes past due.

customer_id: str

event_id: str

event_type: str

invoice_id: str | None

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPastDue'), 'invoice_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionPaused(*, event_id: str = None, event_type: str = 'SubscriptionPaused', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, resumes_at: datetime | None = None)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is paused.

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPaused'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'resumes_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

resumes_at: datetime | None

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionPlanChanged(*, event_id: str = None, event_type: str = 'SubscriptionPlanChanged', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, previous_plan_id: str, new_plan_id: str, is_upgrade: bool)[source]

Bases: DomainEvent

Emitted when subscriptionkore plan is changed.

customer_id: str

event_id: str

event_type: str

is_upgrade: bool

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionPlanChanged'), 'is_upgrade': FieldInfo(annotation=bool, required=True), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'new_plan_id': FieldInfo(annotation=str, required=True), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'previous_plan_id': FieldInfo(annotation=str, required=True), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

new_plan_id: str

occurred_at: datetime

previous_plan_id: str

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionResumed(*, event_id: str = None, event_type: str = 'SubscriptionResumed', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str)[source]

Bases: DomainEvent

Emitted when a subscriptionkore is resumed.

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionResumed'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionTrialEnded(*, event_id: str = None, event_type: str = 'SubscriptionTrialEnded', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, converted: bool)[source]

Bases: DomainEvent

Emitted when a trial ends.

converted: bool

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscriptionkore: Subscription

class subscriptionkore.core.events.SubscriptionTrialStarted(*, event_id: str = None, event_type: str = 'SubscriptionTrialStarted', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, trial_end: datetime)[source]

Bases: DomainEvent

Emitted when a trial starts.

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'customer_id': FieldInfo(annotation=str, required=True), 'event_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'event_type': FieldInfo(annotation=str, required=False, default='SubscriptionTrialStarted'), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'occurred_at': FieldInfo(annotation=datetime, required=False, default_factory=builtin_function_or_method), 'subscriptionkore': FieldInfo(annotation=Subscription, required=True), 'trial_end': FieldInfo(annotation=datetime, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscriptionkore: Subscription

trial_end: datetime

class subscriptionkore.core.events.SubscriptionUpdated(*, event_id: str = None, event_type: str = 'SubscriptionUpdated', occurred_at: datetime = None, metadata: dict[str, Any] = None, subscriptionkore: Subscription, customer_id: str, changed_fields: list[str])[source]

Bases: DomainEvent

Emitted when a subscriptionkore is updated.

changed_fields: list[str]

customer_id: str

event_id: str

event_type: str

metadata: dict[str, Any]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

occurred_at: datetime

subscriptionkore: Subscription

Exceptions¶

Domain exceptions.

exception subscriptionkore.core.exceptions.ConfigurationError(message: str, details: dict[str, Any] | None = None)[source]¶

Bases: SubscriptionKoreError

Base class for configuration errors.

exception subscriptionkore.core.exceptions.DomainError(message: str, details: dict[str, Any] | None = None)[source]¶

Bases: SubscriptionKoreError

Base class for domain errors.

exception subscriptionkore.core.exceptions.DuplicateEntityError(entity_type: str, identifier: str)[source]¶

Bases: DomainError

Raised when attempting to create a duplicate entity.

exception subscriptionkore.core.exceptions.EntitlementError(message: str, details: dict[str, Any] | None = None)[source]¶

Bases: SubscriptionKoreError

Base class for entitlement errors.

exception subscriptionkore.core.exceptions.EntitlementNotFoundError(key: str)[source]¶

Bases: EntitlementError

Raised when an entitlement is not found.

exception subscriptionkore.core.exceptions.EntityNotFoundError(entity_type: str, entity_id: str)[source]¶

Bases: DomainError

Raised when an entity is not found.

exception subscriptionkore.core.exceptions.InvalidConfigurationError(message: str, details: dict[str, Any] | None = None)[source]¶

Bases: ConfigurationError

Raised when configuration is invalid.

exception subscriptionkore.core.exceptions.InvalidStateTransitionError(from_state: SubscriptionStatus, to_state: SubscriptionStatus, reason: str | None = None)[source]¶

Bases: DomainError

Raised when an invalid state transition is attempted.

exception subscriptionkore.core.exceptions.MissingProviderCredentialsError(provider: str, missing_fields: list[str])[source]¶

Bases: ConfigurationError

Raised when provider credentials are missing.

exception subscriptionkore.core.exceptions.ProviderAPIError(message: str, provider: str, status_code: int, provider_message: str | None = None, provider_code: str | None = None)[source]¶

Bases: ProviderError

Raised for non-recoverable API errors from providers.

exception subscriptionkore.core.exceptions.ProviderAuthenticationError(provider: str, message: str | None = None)[source]¶

Bases: ProviderError

Raised when authentication fails. Non-recoverable.

exception subscriptionkore.core.exceptions.ProviderError(message: str, provider: str, details: dict[str, Any] | None = None)[source]¶

Bases: SubscriptionKoreError

Base class for provider errors.

exception subscriptionkore.core.exceptions.ProviderNetworkError(provider: str, original_error: Exception)[source]¶

Bases: ProviderError

Raised for network errors. Recoverable with retry.

exception subscriptionkore.core.exceptions.ProviderRateLimitError(provider: str, retry_after: int | None = None)[source]¶

Bases: ProviderError

Raised when provider rate limit is exceeded. Recoverable with retry.

exception subscriptionkore.core.exceptions.RepositoryError(operation: str, entity_type: str, original_error: Exception)[source]¶

Bases: SubscriptionKoreError

Raised when a repository operation fails.

exception subscriptionkore.core.exceptions.SubscriptionKoreError(message: str, details: dict[str, Any] | None = None)[source]¶

Bases: Exception

Base exception for all subscriptionkore errors.

exception subscriptionkore.core.exceptions.UsageLimitExceededError(entitlement_key: str, limit: int, current_usage: int, requested: int)[source]¶

Bases: EntitlementError

Raised when a usage limit is exceeded.

exception subscriptionkore.core.exceptions.ValidationError(field: str, message: str)[source]¶

Bases: DomainError

Raised when validation fails.

exception subscriptionkore.core.exceptions.WebhookError(message: str, details: dict[str, Any] | None = None)[source]¶

Bases: SubscriptionKoreError

Base class for webhook errors.

exception subscriptionkore.core.exceptions.WebhookPayloadInvalidError(provider: str, reason: str)[source]¶

Bases: WebhookError

Raised when webhook payload cannot be parsed.

exception subscriptionkore.core.exceptions.WebhookProcessingError(event_id: str, reason: str)[source]¶

Bases: WebhookError

Raised when webhook processing fails.

exception subscriptionkore.core.exceptions.WebhookSignatureInvalidError(provider: str)[source]¶

Bases: WebhookError

Raised when webhook signature verification fails.

Adapters¶

Stripe adapter.

class subscriptionkore.adapters.stripe.StripeAdapter(config: StripeConfig)[source]¶

Bases: PaymentProviderPort

Stripe payment provider implementation.

BASE_URL = 'https://api.stripe.com/v1'¶

INVOICE_STATUS_MAP = {'draft': InvoiceStatus.DRAFT, 'open': InvoiceStatus.OPEN, 'paid': InvoiceStatus.PAID, 'uncollectible': InvoiceStatus.UNCOLLECTIBLE, 'void': InvoiceStatus.VOID}¶

STATUS_MAP = {'active': SubscriptionStatus.ACTIVE, 'canceled': SubscriptionStatus.CANCELED, 'incomplete': SubscriptionStatus.INCOMPLETE, 'incomplete_expired': SubscriptionStatus.INCOMPLETE_EXPIRED, 'past_due': SubscriptionStatus.PAST_DUE, 'paused': SubscriptionStatus.PAUSED, 'trialing': SubscriptionStatus.TRIALING, 'unpaid': SubscriptionStatus.UNPAID}¶

async apply_discount(subscriptionkore_provider_ref: ProviderReference, discount: DiscountRequest) → Subscription[source]¶: Apply a discount to a subscriptionkore.

async cancel_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, immediate: bool = False) → Subscription[source]¶: Cancel a subscriptionkore.

property capabilities: ProviderCapabilities¶: Return provider capabilities.

async change_plan(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → Subscription[source]¶: Change subscriptionkore plan (upgrade/downgrade).

async close() → None[source]¶

async create_checkout_session(request: CheckoutRequest, plan_provider_ref: ProviderReference, customer_provider_ref: ProviderReference | None = None) → CheckoutSession[source]¶: Create a checkout session.

async create_customer(customer: Customer) → ProviderReference[source]¶: Create a customer in the provider.

async create_portal_session(customer_provider_ref: ProviderReference, return_url: str) → PortalSession[source]¶: Create a customer portal session.

async create_subscriptionkore(request: CreateSubscriptionRequest, customer_provider_ref: ProviderReference, plan_provider_ref: ProviderReference) → Subscription[source]¶: Create a subscriptionkore.

async delete_customer(provider_ref: ProviderReference) → None[source]¶: Delete a customer from the provider.

async get_customer(provider_ref: ProviderReference) → Customer[source]¶: Get a customer from the provider.

async get_invoice(provider_ref: ProviderReference) → Invoice[source]¶: Get an invoice from the provider.

async get_subscriptionkore(provider_ref: ProviderReference) → Subscription[source]¶: Get a subscriptionkore from the provider.

async get_upcoming_invoice(subscriptionkore_provider_ref: ProviderReference) → Invoice | None[source]¶: Get upcoming invoice for a subscriptionkore.

async list_invoices(customer_provider_ref: ProviderReference, limit: int = 10, starting_after: str | None = None) → list[Invoice][source]¶: List invoices for a customer.

async parse_webhook(payload: bytes, headers: dict[str, str]) → ProviderWebhookEvent[source]¶: Parse Stripe webhook payload.

async pause_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, resumes_at: datetime | None = None) → Subscription[source]¶: Pause a subscriptionkore.

async preview_plan_change(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → ChangePreview[source]¶: Preview plan change costs.

property provider_type: ProviderType¶: Return the provider type.

async remove_discount(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Remove discount from a subscriptionkore.

async resume_subscriptionkore(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Resume a paused subscriptionkore.

async sync_plans(product_provider_ref: ProviderReference) → list[Plan][source]¶: Sync all plans for a product from provider.

async sync_products() → list[Product][source]¶: Sync all products from provider.

async update_customer(customer: Customer) → None[source]¶: Update a customer in the provider.

async update_subscriptionkore(request: UpdateSubscriptionRequest, subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Update a subscriptionkore.

async verify_webhook(payload: bytes, headers: dict[str, str]) → bool[source]¶: Verify Stripe webhook signature.

Chargebee adapter.

class subscriptionkore.adapters.chargebee.ChargebeeAdapter(config: ChargebeeConfig)[source]¶

Bases: PaymentProviderPort

Chargebee payment provider implementation.

Implements the Chargebee API for subscriptionkore management. Reference: https://apidocs.eu.chargebee. com/docs/api/

INVOICE_STATUS_MAP = {'not_paid': InvoiceStatus.OPEN, 'paid': InvoiceStatus.PAID, 'payment_due': InvoiceStatus.OPEN, 'pending': InvoiceStatus.DRAFT, 'posted': InvoiceStatus.OPEN, 'voided': InvoiceStatus.VOID}¶

STATUS_MAP = {'active': SubscriptionStatus.ACTIVE, 'cancelled': SubscriptionStatus.CANCELED, 'future': SubscriptionStatus.INCOMPLETE, 'in_trial': SubscriptionStatus.TRIALING, 'non_renewing': SubscriptionStatus.ACTIVE, 'paused': SubscriptionStatus.PAUSED}¶

async apply_discount(subscriptionkore_provider_ref: ProviderReference, discount: DiscountRequest) → Subscription[source]¶: Apply a discount to a subscriptionkore.

async cancel_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, immediate: bool = False) → Subscription[source]¶: Cancel a subscriptionkore.

property capabilities: ProviderCapabilities¶: Return provider capabilities.

async change_plan(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → Subscription[source]¶: Change subscriptionkore plan (upgrade/downgrade).

async close() → None[source]¶

async create_checkout_session(request: CheckoutRequest, plan_provider_ref: ProviderReference, customer_provider_ref: ProviderReference | None = None) → CheckoutSession[source]¶: Create a checkout session.

async create_customer(customer: Customer) → ProviderReference[source]¶: Create a customer in the provider.

async create_portal_session(customer_provider_ref: ProviderReference, return_url: str) → PortalSession[source]¶: Create a customer portal session.

async create_subscriptionkore(request: CreateSubscriptionRequest, customer_provider_ref: ProviderReference, plan_provider_ref: ProviderReference) → Subscription[source]¶: Create a subscriptionkore.

async delete_customer(provider_ref: ProviderReference) → None[source]¶: Delete a customer from the provider.

async get_customer(provider_ref: ProviderReference) → Customer[source]¶: Get a customer from the provider.

async get_invoice(provider_ref: ProviderReference) → Invoice[source]¶: Get an invoice from the provider.

async get_subscriptionkore(provider_ref: ProviderReference) → Subscription[source]¶: Get a subscriptionkore from the provider.

async get_upcoming_invoice(subscriptionkore_provider_ref: ProviderReference) → Invoice | None[source]¶: Get upcoming invoice for a subscriptionkore.

async list_invoices(customer_provider_ref: ProviderReference, limit: int = 10, starting_after: str | None = None) → list[Invoice][source]¶: List invoices for a customer.

async parse_webhook(payload: bytes, headers: dict[str, str]) → ProviderWebhookEvent[source]¶: Parse Chargebee webhook payload.

async pause_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, resumes_at: datetime | None = None) → Subscription[source]¶: Pause a subscriptionkore.

async preview_plan_change(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → ChangePreview[source]¶: Preview plan change costs.

property provider_type: ProviderType¶: Return the provider type.

async remove_discount(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Remove discount from a subscriptionkore.

async resume_subscriptionkore(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Resume a paused subscriptionkore.

async sync_plans(product_provider_ref: ProviderReference) → list[Plan][source]¶: Sync all plans for a product from provider.

async sync_products() → list[Product][source]¶: Sync all products from provider.

async update_customer(customer: Customer) → None[source]¶: Update a customer in the provider.

async update_subscriptionkore(request: UpdateSubscriptionRequest, subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Update a subscriptionkore.

async verify_webhook(payload: bytes, headers: dict[str, str]) → bool[source]¶: Verify Chargebee webhook using basic auth.

Paddle adapter.

class subscriptionkore.adapters.paddle.PaddleAdapter(config: PaddleConfig)[source]¶

Bases: PaymentProviderPort

Paddle Billing payment provider implementation.

Implements the Paddle Billing API (v1) for subscriptionkore management. Reference: https://developer.paddle.com/api-reference/overview

STATUS_MAP = {'active': SubscriptionStatus.ACTIVE, 'canceled': SubscriptionStatus.CANCELED, 'past_due': SubscriptionStatus.PAST_DUE, 'paused': SubscriptionStatus.PAUSED, 'trialing': SubscriptionStatus.TRIALING}¶

async apply_discount(subscriptionkore_provider_ref: ProviderReference, discount: DiscountRequest) → Subscription[source]¶: Apply a discount to a subscriptionkore.

async cancel_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, immediate: bool = False) → Subscription[source]¶: Cancel a subscriptionkore.

property capabilities: ProviderCapabilities¶: Return provider capabilities.

async change_plan(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → Subscription[source]¶: Change subscriptionkore plan (upgrade/downgrade).

async close() → None[source]¶

async create_checkout_session(request: CheckoutRequest, plan_provider_ref: ProviderReference, customer_provider_ref: ProviderReference | None = None) → CheckoutSession[source]¶: Create a checkout session.

async create_customer(customer: Customer) → ProviderReference[source]¶: Create a customer in the provider.

async create_portal_session(customer_provider_ref: ProviderReference, return_url: str) → PortalSession[source]¶: Create a customer portal session.

async create_subscriptionkore(request: CreateSubscriptionRequest, customer_provider_ref: ProviderReference, plan_provider_ref: ProviderReference) → Subscription[source]¶

Create a subscriptionkore in Paddle.

Note: Paddle subscriptionkores are typically created through checkout. This method creates a subscriptionkore directly via API.

async delete_customer(provider_ref: ProviderReference) → None[source]¶: Delete a customer from the provider.

async get_customer(provider_ref: ProviderReference) → Customer[source]¶: Get a customer from the provider.

async get_invoice(provider_ref: ProviderReference) → Invoice[source]¶: Get an invoice from the provider.

async get_subscriptionkore(provider_ref: ProviderReference) → Subscription[source]¶: Get a subscriptionkore from the provider.

async get_upcoming_invoice(subscriptionkore_provider_ref: ProviderReference) → Invoice | None[source]¶: Get upcoming invoice for a subscriptionkore.

async list_invoices(customer_provider_ref: ProviderReference, limit: int = 10, starting_after: str | None = None) → list[Invoice][source]¶: List invoices for a customer.

async parse_webhook(payload: bytes, headers: dict[str, str]) → ProviderWebhookEvent[source]¶: Parse Paddle webhook payload.

async pause_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, resumes_at: datetime | None = None) → Subscription[source]¶: Pause a subscriptionkore.

async preview_plan_change(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → ChangePreview[source]¶: Preview plan change costs.

property provider_type: ProviderType¶: Return the provider type.

async remove_discount(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Remove discount from a subscriptionkore.

async resume_subscriptionkore(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Resume a paused subscriptionkore.

async sync_plans(product_provider_ref: ProviderReference) → list[Plan][source]¶: Sync all plans for a product from provider.

async sync_products() → list[Product][source]¶: Sync all products from provider.

async update_customer(customer: Customer) → None[source]¶: Update a customer in the provider.

async update_subscriptionkore(request: UpdateSubscriptionRequest, subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Update a subscriptionkore.

async verify_webhook(payload: bytes, headers: dict[str, str]) → bool[source]¶: Verify Paddle webhook signature.

LemonSqueezy adapter.

class subscriptionkore.adapters.lemonsqueezy.LemonSqueezyAdapter(config: LemonSqueezyConfig)[source]¶

Bases: PaymentProviderPort

LemonSqueezy payment provider implementation.

Implements the LemonSqueezy API for subscriptionkore management. Reference: https://docs.lemonsqueezy.com/api

BASE_URL = 'https://api.lemonsqueezy.com/v1'¶

STATUS_MAP = {'active': SubscriptionStatus.ACTIVE, 'cancelled': SubscriptionStatus.CANCELED, 'expired': SubscriptionStatus.EXPIRED, 'on_trial': SubscriptionStatus.TRIALING, 'past_due': SubscriptionStatus.PAST_DUE, 'paused': SubscriptionStatus.PAUSED, 'unpaid': SubscriptionStatus.UNPAID}¶

async apply_discount(subscriptionkore_provider_ref: ProviderReference, discount: DiscountRequest) → Subscription[source]¶: Apply a discount to a subscriptionkore.

async cancel_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, immediate: bool = False) → Subscription[source]¶: Cancel a subscriptionkore.

property capabilities: ProviderCapabilities¶: Return provider capabilities.

async change_plan(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → Subscription[source]¶: Change subscriptionkore plan (upgrade/downgrade).

async close() → None[source]¶

async create_checkout_session(request: CheckoutRequest, plan_provider_ref: ProviderReference, customer_provider_ref: ProviderReference | None = None) → CheckoutSession[source]¶: Create a checkout session.

async create_customer(customer: Customer) → ProviderReference[source]¶: Create a customer in the provider.

async create_portal_session(customer_provider_ref: ProviderReference, return_url: str) → PortalSession[source]¶: Create a customer portal session.

async create_subscriptionkore(request: CreateSubscriptionRequest, customer_provider_ref: ProviderReference, plan_provider_ref: ProviderReference) → Subscription[source]¶: Create a subscriptionkore.

async delete_customer(provider_ref: ProviderReference) → None[source]¶: Delete a customer from the provider.

async get_customer(provider_ref: ProviderReference) → Customer[source]¶: Get a customer from the provider.

async get_invoice(provider_ref: ProviderReference) → Invoice[source]¶: Get an invoice from the provider.

async get_subscriptionkore(provider_ref: ProviderReference) → Subscription[source]¶: Get a subscriptionkore from the provider.

async get_upcoming_invoice(subscriptionkore_provider_ref: ProviderReference) → Invoice | None[source]¶: Get upcoming invoice for a subscriptionkore.

async list_invoices(customer_provider_ref: ProviderReference, limit: int = 10, starting_after: str | None = None) → list[Invoice][source]¶: List invoices for a customer.

async parse_webhook(payload: bytes, headers: dict[str, str]) → ProviderWebhookEvent[source]¶: Parse LemonSqueezy webhook payload.

async pause_subscriptionkore(subscriptionkore_provider_ref: ProviderReference, resumes_at: datetime | None = None) → Subscription[source]¶: Pause a subscriptionkore.

async preview_plan_change(request: ChangePlanRequest, subscriptionkore_provider_ref: ProviderReference, new_plan_provider_ref: ProviderReference) → ChangePreview[source]¶: Preview plan change costs.

property provider_type: ProviderType¶: Return the provider type.

async remove_discount(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Remove discount from a subscriptionkore.

async resume_subscriptionkore(subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Resume a paused subscriptionkore.

async sync_plans(product_provider_ref: ProviderReference) → list[Plan][source]¶: Sync all plans for a product from provider.

async sync_products() → list[Product][source]¶: Sync all products from provider.

async update_customer(customer: Customer) → None[source]¶: Update a customer in the provider.

async update_subscriptionkore(request: UpdateSubscriptionRequest, subscriptionkore_provider_ref: ProviderReference) → Subscription[source]¶: Update a subscriptionkore.

async verify_webhook(payload: bytes, headers: dict[str, str]) → bool[source]¶: Verify LemonSqueezy webhook signature.

Services¶

Application services.

class subscriptionkore.services.CustomerManager(customer_repo: CustomerRepository, provider: PaymentProviderPort, event_bus: EventBusPort)[source]¶

Bases: object

Manages customer lifecycle operations.

Provides a unified interface for customer operations across providers.

async create(external_id: str, email: str, name: str | None = None, tax_info: TaxInfo | None = None, billing_address: Address | None = None, metadata: dict | None = None, sync_to_provider: bool = True) → Customer[source]¶

Create a new customer.

Parameters:

external_id – Your application’s user ID
email – Customer email
name – Customer name
tax_info – Tax information
billing_address – Billing address
metadata – Additional metadata
sync_to_provider – Whether to create in payment provider

Returns:

Created customer

Raises:

DuplicateEntityError – If customer with external_id exists

async delete(customer_id: str, delete_from_provider: bool = True) → bool[source]¶

Delete a customer.

Parameters:

customer_id – Customer ID
delete_from_provider – Whether to delete from payment provider

Returns:

True if deleted

async get(customer_id: str) → Customer[source]¶: Get customer by internal ID.

async get_by_external_id(external_id: str) → Customer[source]¶: Get customer by external (application) ID.

async get_or_create(external_id: str, email: str, name: str | None = None, sync_to_provider: bool = True) → tuple[Customer, bool][source]¶

Get existing customer or create new one.

Returns:: Tuple of (customer, created) where created is True if new

async list(limit: int = 100, offset: int = 0) → list[Customer][source]¶: List customers with pagination.

async sync_to_provider(customer_id: str, provider: ProviderType | None = None) → Customer[source]¶

Sync customer to payment provider.

Creates customer in provider if not exists, or updates if exists.

async update(customer_id: str, email: str | None = None, name: str | None = None, tax_info: TaxInfo | None = None, billing_address: Address | None = None, metadata: dict | None = None, sync_to_provider: bool = True) → Customer[source]¶: Update customer information.

class subscriptionkore.services.EntitlementService(entitlement_repo: EntitlementRepository, override_repo: EntitlementOverrideRepository, subscriptionkore_repo: SubscriptionRepository, plan_repo: PlanRepository, cache: CachePort | None = None, cache_ttl: int = 300)[source]¶

Bases: object

Resolves and manages customer entitlements.

Entitlements are derived from: 1. Customer-level overrides (highest priority) 2. Active subscriptionkore plan entitlements 3. Trial entitlements 4. Default entitlements (lowest priority)

async check(customer_id: str, entitlement_key: str) → CustomerEntitlement[source]¶

Check a specific entitlement for a customer.

Parameters:

customer_id – Customer ID
entitlement_key – Entitlement key to check

Returns:

Resolved CustomerEntitlement

Raises:

EntitlementNotFoundError – If entitlement key doesn’t exist

async check_all(customer_id: str) → list[CustomerEntitlement][source]¶

Get all resolved entitlements for a customer.

Results are cached for performance.

async check_many(customer_id: str, entitlement_keys: list[str]) → dict[str, CustomerEntitlement][source]¶: Check multiple entitlements at once.

async check_within_limit(customer_id: str, entitlement_key: str, current_usage: int, requested: int = 1) → tuple[bool, int | None][source]¶

Check if usage is within limit.

Parameters:

customer_id – Customer ID
entitlement_key – Entitlement key
current_usage – Current usage count
requested – Additional usage being requested

Returns:

Tuple of (within_limit, remaining) remaining is None for unlimited

async enforce_limit(customer_id: str, entitlement_key: str, current_usage: int, requested: int = 1) → int | None[source]¶

Enforce usage limit, raising exception if exceeded.

Parameters:

customer_id – Customer ID
entitlement_key – Entitlement key
current_usage – Current usage count
requested – Additional usage being requested

Returns:

Remaining quota (None for unlimited)

Raises:

UsageLimitExceededError – If limit would be exceeded

async get_limit(customer_id: str, entitlement_key: str) → int | None[source]¶

Get numeric limit for an entitlement.

Returns None for unlimited entitlements.

Grant an entitlement override to a customer.

Parameters:

customer_id – Customer ID
entitlement_key – Entitlement key
value – Override value
expires_at – When override expires (None for permanent)
reason – Reason for override
created_by – Who created the override

Returns:

Created override

async has_access(customer_id: str, entitlement_key: str) → bool[source]¶: Check if customer has access to a feature.

async invalidate(customer_id: str) → None[source]¶: Invalidate entitlement cache for a customer.

async invalidate_all() → None[source]¶: Invalidate all entitlement caches.

async list_overrides(customer_id: str, include_expired: bool = False) → list[EntitlementOverride][source]¶: List all overrides for a customer.

async revoke_override(customer_id: str, entitlement_key: str) → bool[source]¶

Revoke an entitlement override.

Returns:: True if override was deleted

class subscriptionkore.services.SubscriptionManager(subscriptionkore_repo: SubscriptionRepository, customer_repo: CustomerRepository, plan_repo: PlanRepository, provider: PaymentProviderPort, event_bus: EventBusPort)[source]¶

Bases: object

Manages subscriptionkore lifecycle operations.

Provides a unified interface for subscriptionkore operations across providers.

async apply_discount(subscriptionkore_id: str, coupon_code: str) → Subscription[source]¶: Apply a discount to a subscriptionkore.

async cancel(subscriptionkore_id: str, immediate: bool = False, reason: str | None = None) → Subscription[source]¶

Cancel a subscriptionkore.

Parameters:

subscriptionkore_id – Subscription ID
immediate – If True, cancel immediately. If False, cancel at period end.
reason – Optional cancellation reason

Returns:

Updated subscriptionkore

async change_plan(subscriptionkore_id: str, new_plan_id: str, proration_behavior: ProrationBehavior = ProrationBehavior.CREATE_PRORATIONS) → Subscription[source]¶

Change subscriptionkore plan (upgrade/downgrade).

Parameters:

subscriptionkore_id – Subscription ID
new_plan_id – New plan ID
proration_behavior – How to handle proration

Returns:

Updated subscriptionkore

async create(customer_id: str, plan_id: str, quantity: int = 1, trial_period_days: int | None = None, coupon_code: str | None = None, metadata: dict | None = None) → Subscription[source]¶

Create a new subscriptionkore.

Parameters:

customer_id – Internal customer ID
plan_id – Internal plan ID
quantity – Subscription quantity
trial_period_days – Override trial period (None uses plan default)
coupon_code – Optional coupon code to apply
metadata – Additional metadata

Returns:

Created subscriptionkore

Raises:

EntityNotFoundError – If customer or plan not found

async get(subscriptionkore_id: str) → Subscription[source]¶: Get subscriptionkore by ID.

async get_active_by_customer(customer_id: str) → list[Subscription][source]¶: Get active subscriptionkores for a customer.

async get_by_customer(customer_id: str, include_canceled: bool = False) → list[Subscription][source]¶: Get all subscriptionkores for a customer.

async pause(subscriptionkore_id: str, resumes_at: datetime | None = None) → Subscription[source]¶: Pause a subscriptionkore.

async preview_plan_change(subscriptionkore_id: str, new_plan_id: str, proration_behavior: ProrationBehavior = ProrationBehavior.CREATE_PRORATIONS) → ChangePreview[source]¶: Preview costs for a plan change.

async reactivate(subscriptionkore_id: str) → Subscription[source]¶

Reactivate a subscriptionkore that was scheduled to cancel.

Removes the cancel_at_period_end flag.

async remove_discount(subscriptionkore_id: str) → Subscription[source]¶: Remove discount from a subscriptionkore.

async resume(subscriptionkore_id: str) → Subscription[source]¶: Resume a paused subscriptionkore.

async update(subscriptionkore_id: str, quantity: int | None = None, metadata: dict | None = None) → Subscription[source]¶: Update subscriptionkore quantity or metadata.

class subscriptionkore.services.WebhookProcessor(providers: dict[ProviderType, PaymentProviderPort], subscriptionkore_repo: SubscriptionRepository, customer_repo: CustomerRepository, plan_repo: PlanRepository, invoice_repo: InvoiceRepository, payment_event_repo: PaymentEventRepository, processed_event_repo: ProcessedEventRepository, event_bus: EventBusPort, entitlement_service: EntitlementService)[source]¶

Bases: object

Processes webhooks from payment providers.

Responsibilities: - Verify webhook signatures - Parse provider-specific payloads - Normalize to canonical event format - Apply state changes - Emit domain events - Ensure idempotency

async process(provider: str, payload: bytes, headers: dict[str, str]) → WebhookResult[source]¶

Process a webhook from a payment provider.

Parameters:

provider – Provider name (e.g., “stripe”, “paddle”)
payload – Raw webhook payload bytes
headers – HTTP headers from webhook request

Returns:

WebhookResult with processing outcome

Raises:

WebhookSignatureInvalidError – If signature verification fails
WebhookPayloadInvalidError – If payload cannot be parsed
WebhookProcessingError – If processing fails

Integrations¶

Framework integrations.