Skip to content

Model brokers

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,
    /,
    *,
    provider,
    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,
    /,
    *,
    provider: PaymentServiceProvider,
    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(
        provider=provider,
        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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
@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})
    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, /, provider, external_id
)
Source code in components/payment_gateway/subcomponents/banking_documents/models/brokers/sepa_mandate.py 
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,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> SepaMandate:
    sepa_mandate: SepaMandate = session.execute(
        cls.select().filter(
            SepaMandate.provider == provider, 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"]
    )