Skip to content

Model brokers

components.payment_gateway.subcomponents.parties.models.brokers.ca_local_account_financial_instrument

CALocalAccountFinancialInstrumentModelBroker

Bases: BaseModelBroker

Centralized data access for CALocalAccountFinancialInstrument model.

create_ca_local_account_financial_instrument classmethod 

create_ca_local_account_financial_instrument(
    session,
    /,
    *,
    legal_entity_id,
    institution_number,
    transit_number,
    account_number,
)

Create a CA local account financial instrument for the given legal entity.

Parameters:

Name Type Description Default
session Session

SQLAlchemy session to use for DB operations.

 required
legal_entity_id LegalEntityId

ID of the legal entity owning the financial instrument.

 required
institution_number str

Institution number of the bank account.

 required
transit_number str

Transit number of the bank account.

 required
account_number str

Account number of the bank account.

 required
Source code in components/payment_gateway/subcomponents/parties/models/brokers/ca_local_account_financial_instrument.py 
18
19
20
21
22
23
24
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 create_ca_local_account_financial_instrument(
    cls,
    session: Session,
    /,
    *,
    legal_entity_id: LegalEntityId,
    institution_number: str,
    transit_number: str,
    account_number: str,
) -> CALocalAccountFinancialInstrument:
    """Create a CA local account financial instrument for the given legal entity.

    Args:
        session: SQLAlchemy session to use for DB operations.
        legal_entity_id: ID of the legal entity owning the financial instrument.
        institution_number: Institution number of the bank account.
        transit_number: Transit number of the bank account.
        account_number: Account number of the bank account.
    """
    financial_instrument = cls.model(
        legal_entity_id=legal_entity_id,
        ca_local_account_data=CALocalAccountData(
            institution_number=institution_number,
            transit_number=transit_number,
            account_number=account_number,
        ),
    )
    session.add(financial_instrument)
    session.flush()
    session.refresh(financial_instrument)

    return financial_instrument

model class-attribute instance-attribute 

model = CALocalAccountFinancialInstrument

components.payment_gateway.subcomponents.parties.models.brokers.financial_instrument

FinancialInstrumentModelBroker

Bases: BaseModelBroker

Centralized data access for FinancialInstrument model.

get_financial_instrument classmethod 

get_financial_instrument(
    session, /, id, with_legal_entity=False
)
Source code in components/payment_gateway/subcomponents/parties/models/brokers/financial_instrument.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@classmethod
def get_financial_instrument(
    cls,
    session: Session,
    /,
    id: UUID,
    with_legal_entity: bool = False,
) -> FinancialInstrument:
    if with_legal_entity:
        query = cls.select(custom_autoload=cls.autoload | {"legal_entity": True})
    else:
        query = cls.select()
    financial_instrument: FinancialInstrument = session.execute(
        query.filter(FinancialInstrument.id == id)
    ).scalar_one()
    return financial_instrument

model class-attribute instance-attribute 

model = FinancialInstrument

terminate_financial_instrument classmethod 

terminate_financial_instrument(session, /, id)
Source code in components/payment_gateway/subcomponents/parties/models/brokers/financial_instrument.py 
34
35
36
37
38
39
40
41
42
43
44
@classmethod
def terminate_financial_instrument(
    cls,
    session: Session,
    /,
    id: UUID,
) -> FinancialInstrument:
    financial_instrument = cls.get_financial_instrument(session, id)
    financial_instrument.terminated_at = datetime.now()

    return financial_instrument

components.payment_gateway.subcomponents.parties.models.brokers.iban_account_financial_instrument

IBANAccountFinancialInstrumentModelBroker

Bases: BaseModelBroker

Centralized data access for IBANAccountFinancialInstrument model.

create_iban_account_financial_instrument classmethod 

create_iban_account_financial_instrument(
    session,
    /,
    *,
    legal_entity_id,
    iban,
    bank_country_code,
    bic,
)

Create an IBAN account financial instrument for the given legal entity.

Parameters:

Name Type Description Default
session Session

SQLAlchemy session to use for DB operations.

 required
legal_entity_id LegalEntityId

ID of the legal entity owning the financial instrument.

 required
iban str

IBAN of the bank account.

 required
bank_country_code str

Country code of the bank in ISO 3166-1 alpha-2 format.

 required
bic str | None

BIC/SWIFT code of the bank.

 required
Source code in components/payment_gateway/subcomponents/parties/models/brokers/iban_account_financial_instrument.py 
18
19
20
21
22
23
24
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 create_iban_account_financial_instrument(
    cls,
    session: Session,
    /,
    *,
    legal_entity_id: LegalEntityId,
    iban: str,
    bank_country_code: str,
    bic: str | None,
) -> IBANAccountFinancialInstrument:
    """Create an IBAN account financial instrument for the given legal entity.

    Args:
        session: SQLAlchemy session to use for DB operations.
        legal_entity_id: ID of the legal entity owning the financial instrument.
        iban: IBAN of the bank account.
        bank_country_code: Country code of the bank in ISO 3166-1 alpha-2 format.
        bic: BIC/SWIFT code of the bank.
    """
    financial_instrument = cls.model(
        legal_entity_id=legal_entity_id,
        iban_account_data=IBANAccountData(
            iban=iban,
            bank_country_code=bank_country_code,
            bic=bic,
        ),
    )
    session.add(financial_instrument)
    session.flush()
    session.refresh(financial_instrument)

    return financial_instrument

model class-attribute instance-attribute 

model = IBANAccountFinancialInstrument

components.payment_gateway.subcomponents.parties.models.brokers.legal_entity

LegalEntityModelBroker

Bases: BaseModelBroker

Centralized data access for LegalEntity model.

get_legal_entity(session, /, id)

Get a legal entity by ID.

Parameters:

Name Type Description Default
session Session

Database session

 required
id UUID

The legal entity ID

 required

Returns:

Type Description
LegalEntity

The legal entity

Raises:

Type Description
NoResultFound

If no legal entity exists with the given ID
Source code in components/payment_gateway/subcomponents/parties/models/brokers/legal_entity.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@classmethod
def get_legal_entity(
    cls,
    session: Session,
    /,
    id: UUID,
) -> LegalEntity:
    """
    Get a legal entity by ID.

    Args:
        session: Database session
        id: The legal entity ID

    Returns:
        The legal entity

    Raises:
        NoResultFound: If no legal entity exists with the given ID
    """
    legal_entity: LegalEntity = session.execute(
        cls.select().filter(cls.model.id == id)
    ).scalar_one()
    return legal_entity

model class-attribute instance-attribute 

model = LegalEntity

terminate_legal_entity(session, /, id)

Terminate a legal entity by setting its terminated_at timestamp.

Parameters:

Name Type Description Default
session Session

Database session

 required
id UUID

The ID of the legal entity to terminate

 required
Source code in components/payment_gateway/subcomponents/parties/models/brokers/legal_entity.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@classmethod
def terminate_legal_entity(
    cls,
    session: Session,
    /,
    id: UUID,
) -> LegalEntity:
    """
    Terminate a legal entity by setting its terminated_at timestamp.

    Args:
        session: Database session
        id: The ID of the legal entity to terminate
    """
    legal_entity = cls.get_legal_entity(session, id)
    legal_entity.terminated_at = datetime.now()

    return legal_entity

upsert_legal_entity(session, /, *, unique_key, **data)

Create or update a legal entity using the unique_key for deduplication.

If a legal entity with the given unique_key exists, updates only the provided fields (partial update supported). If it doesn't exist, creates a new legal entity with the provided data.

Note: When creating a new entity, the NOT NULL fields (entity_type, legal_name, legal_country_code, address_country_code) must be provided or the database will raise an IntegrityError.

Parameters:

Name Type Description Default
session Session

Database session

 required
unique_key str

The namespaced unique identifier for deduplication

 required
**data Unpack[LegalEntityUpdate]

Fields to set/update (see LegalEntityUpdate)

 {}

Returns:

Type Description
LegalEntity

The created or updated legal entity

Raises:

Type Description
IntegrityError

If creating a new entity without required NOT NULL fields
Source code in components/payment_gateway/subcomponents/parties/models/brokers/legal_entity.py 
 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
 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@classmethod
def upsert_legal_entity(
    cls,
    session: Session,
    /,
    *,
    unique_key: str,
    **data: Unpack[LegalEntityUpdate],
) -> LegalEntity:
    """
    Create or update a legal entity using the unique_key for deduplication.

    If a legal entity with the given unique_key exists, updates only the
    provided fields (partial update supported). If it doesn't exist, creates
    a new legal entity with the provided data.

    Note: When creating a new entity, the NOT NULL fields (entity_type,
    legal_name, legal_country_code, address_country_code) must be provided
    or the database will raise an IntegrityError.

    Args:
        session: Database session
        unique_key: The namespaced unique identifier for deduplication
        **data: Fields to set/update (see LegalEntityUpdate)

    Returns:
        The created or updated legal entity

    Raises:
        IntegrityError: If creating a new entity without required NOT NULL fields
    """

    # Note: even for update scenarios we need to provide all the required
    # fields, else the insert statement will fail before the on conflict do
    # update clause is executed. So we need to get the existing row and use
    #  its values as defaults for the non-nullable columns.
    existing = session.execute(
        cls.select().filter(cls.model.unique_key == unique_key)
    ).scalar_one_or_none()

    # Collect the data for the insert statement
    insert_data = {}
    if existing:
        # Provide explicit values for the non-nullable columns, using the existing row's values as defaults
        for col in [
            "entity_type",
            "legal_name",
            "legal_country_code",
            "address_country_code",
        ]:
            insert_data[col] = getattr(existing, col)
    # Merge the provided data with the existing data
    insert_data.update(data)

    insert_stmt = insert(cls.model).values(
        unique_key=unique_key,
        **insert_data,
    )
    conflict_set = {
        "unique_key": insert_stmt.excluded["unique_key"],
    }
    for key in data.keys():  # Only the provided fields should be updated; this prevents against race conditions in case of concurrent insert/updates
        conflict_set[key] = insert_stmt.excluded[key]
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["unique_key"],  # UNIQUE key / constraint
        set_=conflict_set,
    ).returning(cls.model.id)
    id = session.scalar(upsert_stmt)

    return session.get_one(cls.model, id, populate_existing=True)

LegalEntityUpdate

Bases: TypedDict

Optional fields for updating a LegalEntity via upsert.

All fields are optional to support partial updates. When creating a new entity, the NOT NULL fields (entity_type, legal_name, legal_country_code, address_country_code) must be provided or the database will raise an error.

address_city instance-attribute 

address_city

address_country_code instance-attribute 

address_country_code

address_postal_code instance-attribute 

address_postal_code

address_street instance-attribute 

address_street

address_street_2 instance-attribute 

address_street_2

address_subdivision_code instance-attribute 

address_subdivision_code

entity_type instance-attribute 

entity_type

first_name instance-attribute 

first_name

last_name instance-attribute 

last_name

legal_country_code instance-attribute 

legal_country_code

legal_name instance-attribute 

legal_name