Skip to content

Model brokers

components.payment_gateway.subcomponents.ledgers.models.brokers.ledger

LedgerModelBroker

Bases: BaseModelBroker

autoload class-attribute instance-attribute 

autoload = {'last_entry': True}

create_ledger classmethod 

create_ledger(session, /, *, description, reference=None)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
def create_ledger(
    cls,
    session: Session,
    /,
    *,
    description: str,
    reference: str | None = None,
) -> Ledger:
    ledger = cls.model(
        description=description,
        reference=reference,
    )
    session.add(ledger)
    session.flush()

    return ledger

get_ledger classmethod 

get_ledger(session, /, id, with_entries=False)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@classmethod
def get_ledger(
    cls,
    session: Session,
    /,
    id: UUID,
    with_entries: bool = False,
) -> Ledger:
    if with_entries:
        query = cls.select(custom_autoload=cls.autoload | {"entries": True})
    else:
        query = cls.select()
    ledger: Ledger = session.execute(query.filter(Ledger.id == id)).scalar_one()
    return ledger

list_ledger_ids_by_reference classmethod 

list_ledger_ids_by_reference(session, /, reference)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@classmethod
def list_ledger_ids_by_reference(
    cls,
    session: Session,
    /,
    reference: str,
) -> list[UUID]:
    query = (
        cls.select(custom_autoload={})
        .filter(
            Ledger.reference == reference,
            Ledger.terminated_at.is_(None),
        )
        .options(load_only(Ledger.id))
    )
    ledgers = session.execute(query).scalars().all()
    return [ledger.id for ledger in ledgers]

model class-attribute instance-attribute 

model = Ledger

terminate_ledger classmethod 

terminate_ledger(session, /, id)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger.py 
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def terminate_ledger(
    cls,
    session: Session,
    /,
    id: UUID,
) -> Ledger:
    ledger = cls.get_ledger(session, id)
    ledger.terminated_at = datetime.now()

    return ledger

update_ledger classmethod 

update_ledger(session, /, id, **data)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger.py 
75
76
77
78
79
80
81
82
83
84
85
86
@classmethod
def update_ledger(
    cls,
    session: Session,
    /,
    id: UUID,
    **data: Unpack[LedgerUpdate],
) -> Ledger:
    ledger = cls.get_ledger(session, id)
    ledger.assign(data)

    return ledger

LedgerUpdate

Bases: TypedDict

description instance-attribute 

description

reference instance-attribute 

reference

components.payment_gateway.subcomponents.ledgers.models.brokers.ledger_entry

LedgerEntryModelBroker

Bases: BaseModelBroker

create_ledger_entry classmethod 

create_ledger_entry(
    session,
    /,
    *,
    ledger_id,
    opening_balance,
    amount,
    ending_balance,
    occurred_at,
    description=None,
    reference=None,
    entry_metadata=None,
    external_transaction_id=None,
)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger_entry.py 
 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
@classmethod
def create_ledger_entry(
    cls,
    session: Session,
    /,
    *,
    ledger_id: UUID,
    opening_balance: int,
    amount: int,
    ending_balance: int,
    occurred_at: datetime,
    description: str | None = None,
    reference: str | None = None,
    entry_metadata: dict[str, Any] | None = None,
    external_transaction_id: str | None = None,
) -> LedgerEntry:
    ledger_entry = cls.model(
        ledger_id=ledger_id,
        opening_balance=opening_balance,
        amount=amount,
        ending_balance=ending_balance,
        occurred_at=occurred_at,
        description=description,
        reference=reference,
        entry_metadata=entry_metadata,
        external_transaction_id=external_transaction_id,
    )
    session.add(ledger_entry)
    session.flush()

    # Ensure entry history is updated
    session.refresh(instance=ledger_entry.ledger)

    return ledger_entry

create_ledger_entry_at classmethod 

create_ledger_entry_at(
    session,
    /,
    *,
    ledger_id,
    created_at,
    opening_balance,
    amount,
    ending_balance,
    occurred_at,
    description=None,
    reference=None,
    entry_metadata=None,
    external_transaction_id=None,
)

⚠️ WARNING: This method should be used carefully, as it can break the integrity of the ledger entry chain and should be used with caution.

Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger_entry.py 
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
138
139
140
141
142
143
144
@classmethod
def create_ledger_entry_at(
    cls,
    session: Session,
    /,
    *,
    ledger_id: UUID,
    created_at: datetime,
    opening_balance: int,
    amount: int,
    ending_balance: int,
    occurred_at: datetime,
    description: str | None = None,
    reference: str | None = None,
    entry_metadata: dict[str, Any] | None = None,
    external_transaction_id: str | None = None,
) -> LedgerEntry:
    """
    ⚠️ WARNING: This method should be used carefully, as it can break the integrity
        of the ledger entry chain and should be used with caution.
    """
    ledger_entry = cls.model(
        ledger_id=ledger_id,
        created_at=created_at,
        opening_balance=opening_balance,
        amount=amount,
        ending_balance=ending_balance,
        occurred_at=occurred_at,
        description=description,
        reference=reference,
        entry_metadata=entry_metadata,
        external_transaction_id=external_transaction_id,
    )
    session.add(ledger_entry)
    session.flush()

    # Ensure entry history is updated
    session.refresh(instance=ledger_entry.ledger)

    return ledger_entry

find_ledger_entries_by_reference_prefix classmethod 

find_ledger_entries_by_reference_prefix(
    session, /, ledger_id, reference_prefix
)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger_entry.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@classmethod
def find_ledger_entries_by_reference_prefix(
    cls,
    session: Session,
    /,
    ledger_id: UUID,
    reference_prefix: str,
) -> list[LedgerEntry]:
    ledger_entries = (
        session.execute(
            cls.select().filter(
                LedgerEntry.ledger_id == ledger_id,
                LedgerEntry.reference.like(f"{reference_prefix}%"),
            )
        )
        .scalars()
        .all()
    )
    return list(ledger_entries)

find_ledger_entry_effective_at classmethod 

find_ledger_entry_effective_at(
    session, /, ledger_id, effective_at
)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger_entry.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@classmethod
def find_ledger_entry_effective_at(
    cls,
    session: Session,
    /,
    ledger_id: UUID,
    effective_at: datetime,
) -> LedgerEntry | None:
    ledger_entry: LedgerEntry | None = (
        session.execute(
            cls.select()
            .filter(
                LedgerEntry.ledger_id == ledger_id,
                LedgerEntry.created_at <= effective_at,
            )
            .order_by(LedgerEntry.created_at.desc())
        )
        .scalars()
        .first()
    )
    return ledger_entry

get_ledger_entry classmethod 

get_ledger_entry(session, /, id)
Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger_entry.py 
16
17
18
19
20
21
22
23
24
25
26
@classmethod
def get_ledger_entry(
    cls,
    session: Session,
    /,
    id: UUID,
) -> LedgerEntry:
    ledger_entry: LedgerEntry = session.execute(
        cls.select().filter(LedgerEntry.id == id)
    ).scalar_one()
    return ledger_entry

model class-attribute instance-attribute 

model = LedgerEntry

modify_entry_balance classmethod 

modify_entry_balance(session, /, ledger_id, balance_change)

⚠️ WARNING: This method should be used carefully, as it can break the integrity of the ledger entry chain and should be used with caution.

Source code in components/payment_gateway/subcomponents/ledgers/models/brokers/ledger_entry.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@classmethod
def modify_entry_balance(
    cls,
    session: Session,
    /,
    ledger_id: UUID,
    balance_change: int,
) -> None:
    """
    ⚠️ WARNING: This method should be used carefully, as it can break the integrity
        of the ledger entry chain and should be used with caution.
    """
    ledger_entry = cls.get_ledger_entry(session, ledger_id)
    ledger_entry.opening_balance += balance_change
    ledger_entry.ending_balance += balance_change
    session.flush()

    # Ensure entry history is updated
    session.refresh(instance=ledger_entry.ledger)