components.payment_gateway.public.banking_documents

This module defines the public API for the banking documents subcomponent.

Only business logic classes are exposed here. Basic entities and enums are exposed in separate modules to avoid loading the entire subcomponent with its models and dependencies when they are not needed.

Attributes

PaymentMandateId module-attribute

PaymentMandateId = NewType('PaymentMandateId', UUID)

Classes

PadPaymentMandate dataclass

PadPaymentMandate(
    id,
    unique_key,
    payment_type,
    consent_captured_at,
    status,
    status_history,
)

Bases: PaymentMandate

Public read-only view of a Pre-Authorized Debit (Canada) payment mandate.

PaymentMandate dataclass

PaymentMandate(
    id,
    unique_key,
    payment_type,
    consent_captured_at,
    status,
    status_history,
)

Public read-only view of a PaymentMandate (base — SEPA or PAD).

Attributes

consent_captured_at instance-attribute

consent_captured_at

id instance-attribute

id

payment_type instance-attribute

payment_type

status instance-attribute

status

status_history instance-attribute

status_history

Status log entries, most recent first.

unique_key instance-attribute

unique_key

PaymentMandateActions

PaymentMandateActions()

Actions for payment mandate operations.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/payment_mandate_actions.py

def __init__(self) -> None:
    pass

Functions

create classmethod

create()

Normal factory

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/payment_mandate_actions.py

@classmethod
def create(cls) -> "PaymentMandateActions":
    """Normal factory"""
    return cls()

create_null classmethod

create_null()

Null factory

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/payment_mandate_actions.py

@classmethod
def create_null(cls) -> "PaymentMandateActions":
    """Null factory"""
    return cls()

find_sepa_payment_mandate_by_umr_and_creditor

find_sepa_payment_mandate_by_umr_and_creditor(
    session, /, *, umr, creditor_legal_entity_id
)

Find a SEPA payment mandate by its (umr, creditor) pair.

This pair is unique per SEPA definition.

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
umr	str	Unique Mandate Reference.	required
creditor_legal_entity_id	LegalEntityId	ID of the creditor legal entity.	required

Returns:

Type	Description
SepaPaymentMandate | None	SepaPaymentMandate or None if no mandate matches.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/payment_mandate_actions.py

@obs.api_call()
def find_sepa_payment_mandate_by_umr_and_creditor(
    self,
    session: Session,
    /,
    *,
    umr: str,
    creditor_legal_entity_id: LegalEntityId,
) -> SepaPaymentMandate | None:
    """Find a SEPA payment mandate by its (umr, creditor) pair.

    This pair is unique per SEPA definition.

    Args:
        session: Database session.
        umr: Unique Mandate Reference.
        creditor_legal_entity_id: ID of the creditor legal entity.

    Returns:
        SepaPaymentMandate or None if no mandate matches.
    """
    mandate = SepaPaymentMandateModelBroker.find_by_umr_and_creditor(
        session,
        umr=umr,
        creditor_legal_entity_id=creditor_legal_entity_id,
    )
    if mandate is None:
        return None
    return SepaPaymentMandate(
        id=PaymentMandateId(mandate.id),
        unique_key=mandate.unique_key,
        payment_type=mandate.payment_type,
        consent_captured_at=mandate.consent_captured_at,
        status=mandate.status,
        status_history=tuple(
            PaymentMandateStatusLogEntry(
                status=log.status,
                reason=log.reason,
                created_at=log.created_at,
            )
            for log in mandate.status_history
        ),
        umr=mandate.umr,
        scheme=mandate.scheme,
        valid_until=mandate.valid_until,
    )

set_status

set_status(
    session, /, *, payment_mandate_id, status, reason=None
)

Append a status log entry; no-op if current status already matches.

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
payment_mandate_id	PaymentMandateId	ID of the payment mandate.	required
status	PaymentMandateStatus	Target status.	required
reason	str | None	Optional free-text comment.	None

Raises:

Type	Description
PaymentMandateNotFoundException	If no mandate exists with the given ID.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/payment_mandate_actions.py

@obs.api_call()
def set_status(
    self,
    session: Session,
    /,
    *,
    payment_mandate_id: PaymentMandateId,
    status: PaymentMandateStatus,
    reason: str | None = None,
) -> None:
    """Append a status log entry; no-op if current status already matches.

    Args:
        session: Database session.
        payment_mandate_id: ID of the payment mandate.
        status: Target status.
        reason: Optional free-text comment.

    Raises:
        PaymentMandateNotFoundException: If no mandate exists with the given ID.
    """
    with raise_if_payment_mandate_not_found(payment_mandate_id):
        PaymentMandateModelBroker.set_status(
            session,
            payment_mandate_id=payment_mandate_id,
            status=status,
            reason=reason,
        )

upsert_sepa_payment_mandate

upsert_sepa_payment_mandate(
    session,
    /,
    *,
    unique_key,
    debtor_financial_instrument_id,
    creditor_legal_entity_id,
    payment_type,
    umr,
    scheme,
    valid_until=None,
    consent_captured_at=None,
)

Create-or-update a SEPA payment mandate, keyed by (umr, creditor).

Capturing consent is one-way: once consent_captured_at is set on an existing mandate it is immutable, and status stays ENABLED. The only update permitted is granting consent on a previously PENDING mandate (consent goes None -> not-None, status flips PENDING -> ENABLED). Any other field divergence on update is silently ignored.

On create, status is ENABLED if consent_captured_at is set, else PENDING.

Parameters:

Name	Type	Description	Default
session	Session	Database session.	required
unique_key	str	Namespaced unique identifier {country}:{model}:{id}.	required
debtor_financial_instrument_id	FinancialInstrumentId	ID of the financial instrument to debit from.	required
creditor_legal_entity_id	LegalEntityId	ID of the legal entity authorized to collect.	required
payment_type	PaymentMandatePaymentType	Recurring vs one-off payment authorization.	required
umr	str	Unique Mandate Reference (max 35 chars, Latin only).	required
scheme	SepaPaymentMandateScheme	SEPA scheme (CORE or B2B).	required
valid_until	datetime | None	Mandate validity end date. Defaults to utcnow() + 36 months on create; ignored on update.	None
consent_captured_at	datetime | None	Timestamp of signature or explicit consent.	None

Returns:

Name	Type	Description
PaymentMandateId	PaymentMandateId	ID of the created or updated mandate.

Raises:

Type	Description
FinancialInstrumentNotFoundException	If the debtor FI does not exist (create only).
FinancialInstrumentTerminatedException	If the debtor FI is terminated (create only).
LegalEntityNotFoundException	If the creditor LE does not exist (create only).
LegalEntityTerminatedException	If the creditor LE is terminated (create only).

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/payment_mandate_actions.py

@obs.api_call()
def upsert_sepa_payment_mandate(
    self,
    session: Session,
    /,
    *,
    unique_key: str,
    debtor_financial_instrument_id: FinancialInstrumentId,
    creditor_legal_entity_id: LegalEntityId,
    payment_type: PaymentMandatePaymentType,
    umr: str,
    scheme: SepaPaymentMandateScheme,
    valid_until: datetime | None = None,
    consent_captured_at: datetime | None = None,
) -> PaymentMandateId:
    """Create-or-update a SEPA payment mandate, keyed by `(umr, creditor)`.

    Capturing consent is one-way: once `consent_captured_at` is set on an
    existing mandate it is immutable, and status stays ENABLED. The only
    update permitted is granting consent on a previously PENDING mandate
    (consent goes None -> not-None, status flips PENDING -> ENABLED).
    Any other field divergence on update is silently ignored.

    On create, status is ENABLED if `consent_captured_at` is set, else PENDING.

    Args:
        session: Database session.
        unique_key: Namespaced unique identifier `{country}:{model}:{id}`.
        debtor_financial_instrument_id: ID of the financial instrument to debit from.
        creditor_legal_entity_id: ID of the legal entity authorized to collect.
        payment_type: Recurring vs one-off payment authorization.
        umr: Unique Mandate Reference (max 35 chars, Latin only).
        scheme: SEPA scheme (CORE or B2B).
        valid_until: Mandate validity end date. Defaults to
            `utcnow() + 36 months` on create; ignored on update.
        consent_captured_at: Timestamp of signature or explicit consent.

    Returns:
        PaymentMandateId: ID of the created or updated mandate.

    Raises:
        FinancialInstrumentNotFoundException: If the debtor FI does not exist (create only).
        FinancialInstrumentTerminatedException: If the debtor FI is terminated (create only).
        LegalEntityNotFoundException: If the creditor LE does not exist (create only).
        LegalEntityTerminatedException: If the creditor LE is terminated (create only).
    """
    existing = SepaPaymentMandateModelBroker.find_by_umr_and_creditor(
        session,
        umr=umr,
        creditor_legal_entity_id=creditor_legal_entity_id,
    )
    if existing is not None:
        # consent_captured_at is one-way (PENDING -> ENABLED). Once set it is
        # immutable; ignore later attempts to overwrite or clear it.
        if existing.consent_captured_at is None and consent_captured_at is not None:
            existing.consent_captured_at = consent_captured_at
            PaymentMandateModelBroker.set_status(
                session,
                payment_mandate_id=PaymentMandateId(existing.id),
                status=PaymentMandateStatus.ENABLED,
            )
        return PaymentMandateId(existing.id)

    with raise_if_financial_instrument_not_found(debtor_financial_instrument_id):
        debtor_financial_instrument = (
            FinancialInstrumentModelBroker.get_financial_instrument(
                session, id=debtor_financial_instrument_id
            )
        )
    raise_on_terminated_financial_instrument(debtor_financial_instrument)

    with raise_if_legal_entity_not_found(creditor_legal_entity_id):
        creditor_legal_entity = LegalEntityModelBroker.get_legal_entity(
            session, id=creditor_legal_entity_id
        )
    raise_on_terminated_legal_entity(creditor_legal_entity)

    status = (
        PaymentMandateStatus.ENABLED
        if consent_captured_at is not None
        else PaymentMandateStatus.PENDING
    )

    mandate = SepaPaymentMandateModelBroker.create_sepa_payment_mandate(
        session,
        unique_key=unique_key,
        debtor_financial_instrument_id=debtor_financial_instrument_id,
        creditor_legal_entity_id=creditor_legal_entity_id,
        payment_type=payment_type,
        umr=umr,
        scheme=scheme,
        status=status,
        valid_until=valid_until,
        consent_captured_at=consent_captured_at,
    )
    return PaymentMandateId(mandate.id)

SepaMandateActions

Sepa Mandate Actions

Functions

declare_sepa_mandate

declare_sepa_mandate(
    session,
    /,
    workspace_key,
    external_id,
    account_holder_id,
    sepa_creditor_identifier,
    debtor_name,
    debtor_iban,
    debtor_country,
    unique_mandate_reference,
    issued_at,
    status,
)

Declare a SEPA mandate.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/sepa_mandate_actions.py

@obs.api_call()
def declare_sepa_mandate(
    self,
    session: Session,
    /,
    workspace_key: str,
    external_id: str,
    account_holder_id: AccountHolderId,
    sepa_creditor_identifier: str,
    debtor_name: str,
    debtor_iban: str,
    debtor_country: str,
    unique_mandate_reference: str,
    issued_at: datetime,
    status: SepaMandateStatus,
) -> SepaMandateId:
    """
    Declare a SEPA mandate.
    """
    sepa_mandate = SepaMandateModelBroker.create_sepa_mandate(
        session,
        workspace_key=workspace_key,
        external_id=external_id,
        status=status,
        account_holder_id=account_holder_id,
        sepa_creditor_identifier=sepa_creditor_identifier,
        debtor_name=debtor_name,
        debtor_iban=debtor_iban,
        debtor_country=debtor_country,
        unique_mandate_reference=unique_mandate_reference,
        issued_at=issued_at,
    )
    return SepaMandateId(sepa_mandate.id)

update_sepa_mandate_status

update_sepa_mandate_status(
    session, /, sepa_mandate_id, status
)

Update the status of a specific SEPA mandate.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/actions/sepa_mandate_actions.py

@obs.api_call()
def update_sepa_mandate_status(
    self,
    session: Session,
    /,
    sepa_mandate_id: SepaMandateId,
    status: SepaMandateStatus,
) -> None:
    """
    Update the status of a specific SEPA mandate.
    """
    with raise_if_sepa_mandate_not_found(sepa_mandate_id):
        SepaMandateModelBroker.set_sepa_mandate_status(
            session,
            id=sepa_mandate_id,
            status=status,
        )

SepaMandateQueries

Sepa Mandate Queries

Functions

get_sepa_mandate

get_sepa_mandate(session, /, id)

Get a SEPA mandate by ID.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/queries/sepa_mandate_queries.py

@obs.api_call()
def get_sepa_mandate(
    self,
    session: Session,
    /,
    id: SepaMandateId,
) -> SepaMandate:
    """
    Get a SEPA mandate by ID.
    """
    with raise_if_sepa_mandate_not_found(id):
        sepa_mandate = SepaMandateModelBroker.get_sepa_mandate(session, id=id)

    return _to_dataclass(sepa_mandate=sepa_mandate)

get_sepa_mandate_by_external_id

get_sepa_mandate_by_external_id(
    session, /, workspace_key, external_id
)

Get a SEPA mandate by its external ID.

Source code in components/payment_gateway/subcomponents/banking_documents/protected/business_logic/queries/sepa_mandate_queries.py

@obs.api_call()
def get_sepa_mandate_by_external_id(
    self,
    session: Session,
    /,
    workspace_key: str,
    external_id: str,
) -> SepaMandate:
    """
    Get a SEPA mandate by its external ID.
    """
    with raise_if_sepa_mandate_not_found_for_external_id(external_id):
        sepa_mandate = SepaMandateModelBroker.get_sepa_mandate_by_external_id(
            session,
            workspace_key=workspace_key,
            external_id=external_id,
        )

    return _to_dataclass(sepa_mandate=sepa_mandate)

SepaPaymentMandate dataclass

SepaPaymentMandate(
    id,
    unique_key,
    payment_type,
    consent_captured_at,
    status,
    status_history,
    umr,
    scheme,
    valid_until,
)

Bases: PaymentMandate

Public read-only view of a SEPA payment mandate.

Attributes

scheme instance-attribute

scheme

umr instance-attribute

umr

valid_until instance-attribute

valid_until