Skip to content

Model brokers

components.payment_gateway.subcomponents.accounts.models.brokers.account

AccountModelBroker

Bases: BaseModelBroker

autoload class-attribute instance-attribute 

autoload = {'current_status_log': True}

create_account classmethod 

create_account(
    session,
    /,
    *,
    provider,
    external_id,
    description,
    reference=None,
    account_holder_id=None,
    status=AccountStatus.inactive,
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
 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
@classmethod
def create_account(
    cls,
    session: Session,
    /,
    *,
    provider: PaymentServiceProvider,
    external_id: str,
    description: str,
    reference: str | None = None,
    account_holder_id: UUID | None = None,
    status: AccountStatus = AccountStatus.inactive,
) -> Account:
    account = cls.model(
        provider=provider,
        external_id=external_id,
        description=description,
        reference=reference,
        account_holder_id=account_holder_id,
    )
    status_log = AccountStatusLog(account=account, status=status)
    session.add(account)
    session.add(status_log)
    session.flush()

    return account

get_account classmethod 

get_account(session, /, id, with_status_history=False)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@classmethod
def get_account(
    cls,
    session: Session,
    /,
    id: UUID,
    with_status_history: bool = False,
) -> Account:
    if with_status_history:
        query = cls.select(custom_autoload=cls.autoload | {"status_history": True})
    else:
        query = cls.select()
    account: Account = session.execute(query.filter(Account.id == id)).scalar_one()
    return account

get_account_id_by_external_id classmethod 

get_account_id_by_external_id(
    session, /, provider, external_id
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@classmethod
def get_account_id_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> UUID:
    account: Account = session.execute(
        cls.select(custom_autoload={})
        .filter(Account.provider == provider, Account.external_id == external_id)
        .options(load_only(Account.id))
    ).scalar_one()
    return account.id

list_account_ids_for_account_holder classmethod 

list_account_ids_for_account_holder(
    session, /, account_holder_id, with_terminated=False
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@classmethod
def list_account_ids_for_account_holder(
    cls,
    session: Session,
    /,
    account_holder_id: UUID,
    with_terminated: bool = False,
) -> list[UUID]:
    query = (
        cls.select(custom_autoload={})
        .filter(Account.account_holder_id == account_holder_id)
        .options(load_only(Account.id))
    )
    if not with_terminated:
        query = query.filter(Account.terminated_at.is_(None))
    accounts = session.execute(query).scalars().all()
    return [account.id for account in accounts]

list_accounts_for_account_holder classmethod 

list_accounts_for_account_holder(
    session, /, account_holder_id, with_terminated=False
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
78
79
80
81
82
83
84
85
86
87
88
89
@classmethod
def list_accounts_for_account_holder(
    cls,
    session: Session,
    /,
    account_holder_id: UUID,
    with_terminated: bool = False,
) -> list[Account]:
    query = cls.select().filter(Account.account_holder_id == account_holder_id)
    if not with_terminated:
        query = query.filter(Account.terminated_at.is_(None))
    return list(session.execute(query).scalars().all())

model class-attribute instance-attribute 

model = Account

set_account_status classmethod 

set_account_status(session, /, id, status)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def set_account_status(
    cls,
    session: Session,
    /,
    id: UUID,
    status: AccountStatus,
) -> None:
    account = cls.get_account(session, id)
    if account.status == status:
        return
    status_log = AccountStatusLog(account=account, status=status)
    session.add(status_log)
    session.refresh(account)

terminate_account classmethod 

terminate_account(session, /, id)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
146
147
148
149
150
151
152
153
154
155
156
@classmethod
def terminate_account(
    cls,
    session: Session,
    /,
    id: UUID,
) -> Account:
    account = cls.get_account(session, id)
    account.terminated_at = datetime.now()

    return account

update_account classmethod 

update_account(session, /, id, **data)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account.py 
118
119
120
121
122
123
124
125
126
127
128
129
@classmethod
def update_account(
    cls,
    session: Session,
    /,
    id: UUID,
    **data: Unpack[AccountUpdate],
) -> Account:
    account = cls.get_account(session, id)
    account.assign(data)

    return account

AccountUpdate

Bases: TypedDict

account_holder_id instance-attribute 

account_holder_id

description instance-attribute 

description

reference instance-attribute 

reference

components.payment_gateway.subcomponents.accounts.models.brokers.account_holder

AccountHolderModelBroker

Bases: BaseModelBroker

create_account_holder classmethod 

create_account_holder(
    session,
    /,
    *,
    provider,
    external_id,
    description,
    reference=None,
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account_holder.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@classmethod
def create_account_holder(
    cls,
    session: Session,
    /,
    *,
    provider: PaymentServiceProvider,
    external_id: str,
    description: str,
    reference: str | None = None,
) -> AccountHolder:
    account_holder = cls.model(
        provider=provider,
        external_id=external_id,
        description=description,
        reference=reference,
    )
    session.add(account_holder)
    session.flush()

    return account_holder

get_account_holder classmethod 

get_account_holder(session, /, id)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account_holder.py 
22
23
24
25
26
27
28
29
30
31
32
@classmethod
def get_account_holder(
    cls,
    session: Session,
    /,
    id: UUID,
) -> AccountHolder:
    account_holder: AccountHolder = session.execute(
        cls.select().filter(AccountHolder.id == id)
    ).scalar_one()
    return account_holder

get_account_holder_id_by_external_id classmethod 

get_account_holder_id_by_external_id(
    session, /, provider, external_id
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account_holder.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def get_account_holder_id_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> UUID:
    account_holder: AccountHolder = session.execute(
        cls.select(custom_autoload={})
        .filter(
            AccountHolder.provider == provider,
            AccountHolder.external_id == external_id,
        )
        .options(load_only(AccountHolder.id))
    ).scalar_one()
    return account_holder.id

model class-attribute instance-attribute 

model = AccountHolder

terminate_account_holder classmethod 

terminate_account_holder(session, /, id)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account_holder.py 
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def terminate_account_holder(
    cls,
    session: Session,
    /,
    id: UUID,
) -> AccountHolder:
    account_holder = cls.get_account_holder(session, id)
    account_holder.terminated_at = datetime.now()

    return account_holder

update_account_holder classmethod 

update_account_holder(session, /, id, **data)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/account_holder.py 
74
75
76
77
78
79
80
81
82
83
84
85
@classmethod
def update_account_holder(
    cls,
    session: Session,
    /,
    id: UUID,
    **data: Unpack[AccountHolderUpdate],
) -> AccountHolder:
    account_holder = cls.get_account_holder(session, id)
    account_holder.assign(data)

    return account_holder

AccountHolderUpdate

Bases: TypedDict

description instance-attribute 

description

reference instance-attribute 

reference

components.payment_gateway.subcomponents.accounts.models.brokers.sepa_beneficiary

SepaBeneficiaryModelBroker

Bases: BaseModelBroker

autoload class-attribute instance-attribute 

autoload = {'current_status_log': True}

create_sepa_beneficiary classmethod 

create_sepa_beneficiary(
    session,
    /,
    *,
    provider,
    external_id,
    name,
    iban,
    issued_at,
    account_id,
    status=SepaBeneficiaryStatus.enabled,
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/sepa_beneficiary.py 
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
@classmethod
def create_sepa_beneficiary(
    cls,
    session: Session,
    /,
    *,
    provider: PaymentServiceProvider,
    external_id: str,
    name: str,
    iban: str,
    issued_at: datetime,
    account_id: UUID,
    status: SepaBeneficiaryStatus = SepaBeneficiaryStatus.enabled,
) -> SepaBeneficiary:
    sepa_beneficiary = cls.model(
        provider=provider,
        external_id=external_id,
        name=name,
        iban=iban,
        issued_at=issued_at,
        account_id=account_id,
    )
    status_log = SepaBeneficiaryStatusLog(
        sepa_beneficiary=sepa_beneficiary, status=status
    )

    session.add(sepa_beneficiary)
    session.add(status_log)
    session.flush()

    return sepa_beneficiary

find_sepa_beneficiary_by_external_id classmethod 

find_sepa_beneficiary_by_external_id(
    session, /, provider, external_id
)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/sepa_beneficiary.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@classmethod
def find_sepa_beneficiary_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> SepaBeneficiary | None:
    return session.execute(
        cls.select().filter(
            SepaBeneficiary.provider == provider,
            SepaBeneficiary.external_id == external_id,
        )
    ).scalar_one_or_none()

get_sepa_beneficiary classmethod 

get_sepa_beneficiary(session, /, id)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/sepa_beneficiary.py 
24
25
26
27
28
29
30
31
32
33
34
@classmethod
def get_sepa_beneficiary(
    cls,
    session: Session,
    /,
    id: UUID,
) -> SepaBeneficiary:
    return cast(
        "SepaBeneficiary",
        session.execute(cls.select().filter(SepaBeneficiary.id == id)).scalar_one(),
    )

model class-attribute instance-attribute 

model = SepaBeneficiary

set_sepa_beneficiary_status classmethod 

set_sepa_beneficiary_status(session, /, id, status)
Source code in components/payment_gateway/subcomponents/accounts/models/brokers/sepa_beneficiary.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@classmethod
def set_sepa_beneficiary_status(
    cls,
    session: Session,
    /,
    id: UUID,
    status: SepaBeneficiaryStatus,
) -> None:
    sepa_beneficiary = cls.get_sepa_beneficiary(session, id)
    if sepa_beneficiary.status == status:
        return

    status_log = SepaBeneficiaryStatusLog(
        sepa_beneficiary=sepa_beneficiary, status=status
    )
    session.add(status_log)

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