Skip to content

Model brokers

components.payment_gateway.subcomponents.banking_documents.models.brokers.payment_mandate

PaymentMandateModelBroker

Bases: BaseModelBroker

Centralized data access for PaymentMandate model (polymorphic parent).

get_payment_mandate classmethod 

get_payment_mandate(session, /, id)

Get a payment mandate by ID.

Returns the concrete subtype (SepaPaymentMandate, PadPaymentMandate, ...) via SQLAlchemy joined-inheritance polymorphic loading.

Parameters:

Name Type Description Default
session Session

Database session.

 required
id UUID

The payment mandate ID.

 required

Returns:

Type Description
PaymentMandate

The payment mandate.

Raises:

Type Description
NoResultFound

If no mandate exists with the given ID.
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/payment_mandate.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def get_payment_mandate(
    cls,
    session: Session,
    /,
    id: UUID,
) -> PaymentMandate:
    """Get a payment mandate by ID.

    Returns the concrete subtype (SepaPaymentMandate, PadPaymentMandate, ...)
    via SQLAlchemy joined-inheritance polymorphic loading.

    Args:
        session: Database session.
        id: The payment mandate ID.

    Returns:
        The payment mandate.

    Raises:
        NoResultFound: If no mandate exists with the given ID.
    """
    mandate: PaymentMandate = session.execute(
        cls.select().filter(PaymentMandate.id == id)
    ).scalar_one()
    return mandate

model class-attribute instance-attribute 

model = PaymentMandate

set_status classmethod 

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
NoResultFound

If no mandate exists with the given ID.
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/payment_mandate.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@classmethod
def set_status(
    cls,
    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:
        NoResultFound: If no mandate exists with the given ID.
    """
    mandate = cls.get_payment_mandate(session, id=payment_mandate_id)
    if mandate.status == status:
        return

    status_log = PaymentMandateStatusLog(
        payment_mandate=mandate,
        status=status,
        reason=reason,
    )
    session.add(status_log)

    session.refresh(
        mandate, attribute_names=["status_history", "current_status_log"]
    )

components.payment_gateway.subcomponents.banking_documents.models.brokers.sepa_mandate

SepaMandateModelBroker

Bases: BaseModelBroker

autoload class-attribute instance-attribute 

autoload = {'current_status_log': True}

create_sepa_mandate classmethod 

create_sepa_mandate(
    session,
    /,
    *,
    workspace_key,
    external_id,
    sepa_creditor_identifier,
    debtor_name,
    debtor_iban,
    debtor_country,
    unique_mandate_reference,
    issued_at,
    account_holder_id=None,
    status,
)
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_mandate.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def create_sepa_mandate(
    cls,
    session: Session,
    /,
    *,
    workspace_key: str,
    external_id: str,
    sepa_creditor_identifier: str,
    debtor_name: str,
    debtor_iban: str,
    debtor_country: str,
    unique_mandate_reference: str,
    issued_at: datetime,
    account_holder_id: UUID | None = None,
    status: SepaMandateStatus,
) -> SepaMandate:
    sepa_mandate = cls.model(
        workspace_key=workspace_key,
        external_id=external_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,
        account_holder_id=account_holder_id,
    )
    session.add(sepa_mandate)

    status_log = SepaMandateStatusLog(
        sepa_mandate=sepa_mandate,
        status=status,
    )
    session.add(status_log)

    session.flush()

    return sepa_mandate

get_sepa_mandate classmethod 

get_sepa_mandate(session, /, id, with_status_history=True)
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_mandate.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@classmethod
def get_sepa_mandate(
    cls,
    session: Session,
    /,
    id: UUID,
    with_status_history: bool = True,  # TODO @frederic.bonnet 2025-11-01: default to False but need to update tests
) -> SepaMandate:
    if with_status_history:
        query = cls.select(custom_autoload=cls.autoload | {"status_history": True})  # pyrefly: ignore [bad-argument-type]
    else:
        query = cls.select()
    sepa_mandate: SepaMandate = session.execute(
        query.filter(SepaMandate.id == id)
    ).scalar_one()
    return sepa_mandate

get_sepa_mandate_by_external_id classmethod 

get_sepa_mandate_by_external_id(
    session, /, workspace_key, external_id
)
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_mandate.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def get_sepa_mandate_by_external_id(
    cls,
    session: Session,
    /,
    workspace_key: str,
    external_id: str,
) -> SepaMandate:
    sepa_mandate: SepaMandate = session.execute(
        cls.select().filter(
            SepaMandate.workspace_key == workspace_key,
            SepaMandate.external_id == external_id,
        )
    ).scalar_one()
    return sepa_mandate

model class-attribute instance-attribute 

model = SepaMandate

set_sepa_mandate_status classmethod 

set_sepa_mandate_status(session, /, id, status)
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_mandate.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@classmethod
def set_sepa_mandate_status(
    cls,
    session: Session,
    /,
    id: UUID,
    status: SepaMandateStatus,
) -> None:
    sepa_mandate = cls.get_sepa_mandate(session, id)
    if sepa_mandate.status == status:
        return

    status_log = SepaMandateStatusLog(
        sepa_mandate=sepa_mandate,
        status=status,
    )
    session.add(status_log)

    session.refresh(
        sepa_mandate, attribute_names=["status_history", "current_status_log"]
    )

components.payment_gateway.subcomponents.banking_documents.models.brokers.sepa_payment_mandate

SEPA_MANDATE_VALIDITY_MONTHS module-attribute 

SEPA_MANDATE_VALIDITY_MONTHS = 36

SEPA mandates expire if not used for 36 months (SEPA scheme rule).

SepaPaymentMandateModelBroker

Bases: BaseModelBroker

Centralized data access for SepaPaymentMandate model.

create_sepa_payment_mandate classmethod 

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

Create a SEPA payment mandate together with its initial status log.

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
status PaymentMandateStatus

Initial lifecycle status (logged via PaymentMandateStatusLog).

 required
valid_until datetime | None

Mandate validity end date. Defaults to utcnow() + 36 months if not provided.

 None
consent_captured_at datetime | None

Timestamp of signature or explicit consent.

 None

Returns:

Type Description
SepaPaymentMandate

The newly created SEPA payment mandate.
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_payment_mandate.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@classmethod
def create_sepa_payment_mandate(
    cls,
    session: Session,
    /,
    *,
    unique_key: str,
    debtor_financial_instrument_id: FinancialInstrumentId,
    creditor_legal_entity_id: LegalEntityId,
    payment_type: PaymentMandatePaymentType,
    umr: str,
    scheme: SepaPaymentMandateScheme,
    status: PaymentMandateStatus,
    valid_until: datetime | None = None,
    consent_captured_at: datetime | None = None,
) -> SepaPaymentMandate:
    """Create a SEPA payment mandate together with its initial status log.

    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).
        status: Initial lifecycle status (logged via PaymentMandateStatusLog).
        valid_until: Mandate validity end date. Defaults to
            `utcnow() + 36 months` if not provided.
        consent_captured_at: Timestamp of signature or explicit consent.

    Returns:
        The newly created SEPA payment mandate.
    """
    if valid_until is None:
        valid_until = utcnow() + relativedelta(months=SEPA_MANDATE_VALIDITY_MONTHS)

    mandate = cls.model(
        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,
        valid_until=valid_until,
        consent_captured_at=consent_captured_at,
    )
    session.add(mandate)

    status_log = PaymentMandateStatusLog(
        payment_mandate=mandate,
        status=status,
    )
    session.add(status_log)

    session.flush()
    session.refresh(mandate)

    return mandate

find_by_umr_and_creditor classmethod 

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

Find a SEPA payment mandate by its (umr, creditor) pair. This 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

The mandate, or None if not found.
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_payment_mandate.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@classmethod
def find_by_umr_and_creditor(
    cls,
    session: Session,
    /,
    *,
    umr: str,
    creditor_legal_entity_id: LegalEntityId,
) -> SepaPaymentMandate | None:
    """Find a SEPA payment mandate by its (umr, creditor) pair.
    This is unique per SEPA definition

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

    Returns:
        The mandate, or None if not found.
    """
    return session.scalar(
        select(cls.model).where(
            cls.model.umr == umr,
            cls.model.creditor_legal_entity_id == creditor_legal_entity_id,
        )
    )

model class-attribute instance-attribute 

model = SepaPaymentMandate