Skip to content

Model brokers

components.payment_gateway.subcomponents.authorizations.models.brokers.authorization_request

AuthorizationRequestModelBroker

Bases: BaseModelBroker

SQL model broker for AuthorizationRequest model.

See: https://www.notion.so/alaninsurance/SQL-Model-Brokers-90876b7e739c4195a6a2a4106139ed38?pvs=4 ⧉

find_authorization_request_by_external_id classmethod 

find_authorization_request_by_external_id(
    session, /, provider, external_id
)

Get an authorization request by its provider-specific external ID.

Parameters:

Name Type Description Default
provider PaymentServiceProvider

The provider of the authorization request.

 required
external_id str

The external ID of the authorization request.

 required

Returns:

Type Description
AuthorizationRequest | None

The authorization request if found, else None.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/authorization_request.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def find_authorization_request_by_external_id(
    cls,
    session: Session,
    /,
    provider: PaymentServiceProvider,
    external_id: str,
) -> AuthorizationRequest | None:
    """Get an authorization request by its provider-specific external ID.

    Args:
        provider: The provider of the authorization request.
        external_id: The external ID of the authorization request.

    Returns:
        The authorization request if found, else None.
    """
    return session.execute(
        cls.select().filter(
            and_(
                AuthorizationRequest.provider == provider,
                AuthorizationRequest.external_id == external_id,
            )
        )
    ).scalar_one_or_none()

model class-attribute instance-attribute 

model = AuthorizationRequest

upsert_authorization_request classmethod 

upsert_authorization_request(
    session,
    /,
    *,
    status,
    authorization_metadata,
    provider,
    external_id,
)

Upsert an authorization request.

If an authorization request with the same provider and external ID does not exist, a new one is created and returned. If it does exist, the status is updated and the old status is returned along with the updated authorization request (the other fields are left untouched).

Possible status transitions in this case are:

  • if the old status is active and the new status is approved or declined, the new status is set to complete
  • if the old status is approved or declined and the new status is active, the new status is set to complete
  • otherwise, the status is unchanged
Note

Although the upsert itself is atomic, the status update is not because of limitations with on_conflict_do_update (we cannot update the status conditionally based on the old status as required by the transition rules expressed above). However this is OK because our transition rules are idempotent, so there's no risk of ABA problem here.

Which brings another issue about knowing whether a create or update occurred. on_conflict_do_update provides no direct way of knowing that, other than comparing the before and after values. We cannot know the old value if it's also updated either, so we have to do it in two steps and perform the transition programmatically.

A robust way of telling between create and update cases is to force the model's UUIID at creation time and comparing it with the actual value returned by the statement. If they are the same, it means that the row was just created. If they are different, it means that the row was updated, because this value will never change.

Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/authorization_request.py 
 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
 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
138
@classmethod
def upsert_authorization_request(
    cls,
    session: Session,
    /,
    *,
    status: AuthorizationRequestStatus,
    authorization_metadata: dict[str, Any],
    provider: PaymentServiceProvider,
    external_id: str,
) -> tuple[AuthorizationRequest, AuthorizationRequestStatus | None]:
    """Upsert an authorization request.

    If an authorization request with the same provider and external ID does
    not exist, a new one is created and returned. If it does exist, the
    status is updated and the old status is returned along with the updated
    authorization request (the other fields are left untouched).

    Possible status transitions in this case are:

    - if the old status is active and the new status is approved or declined,
      the new status is set to complete
    - if the old status is approved or declined and the new status is active,
      the new status is set to complete
    - otherwise, the status is unchanged

    Note:
        Although the upsert itself is atomic, the status update is not
        because of limitations with `on_conflict_do_update` (we cannot
        update the status conditionally based on the old status as required
        by the transition rules expressed above). However this is OK because
        our transition rules are idempotent, so there's no risk of ABA
        problem here.

        Which brings another issue about knowing whether a create or update
        occurred. `on_conflict_do_update` provides no direct way of knowing
        that, other than comparing the before and after values. We cannot
        know the old value if it's also updated either, so we have to do it
        in two steps and perform the transition programmatically.

        A robust way of telling between create and update cases is to force
        the model's UUIID at creation time and comparing it with the actual
        value returned by the statement. If they are the same, it means
        that the row was just created. If they are different, it means that
        the row was updated, because this value will never change.
    """
    create_id = uuid.uuid4()
    insert_stmt = insert(cls.model).values(
        id=create_id,
        status=status,
        authorization_metadata=authorization_metadata,
        provider=provider,
        external_id=external_id,
    )
    upsert_stmt = insert_stmt.on_conflict_do_update(
        constraint="authorization_request__unique_external_id_per_provider",
        set_={"updated_at": func.now()},
    ).returning(cls.model.id)

    result_id = session.scalar(upsert_stmt)
    authorization_request = session.execute(
        cls.select().filter(cls.model.id == result_id)
    ).scalar_one()

    if create_id == result_id:
        # The row was just created, so the old status is none
        old_status = None
    else:
        # The row was updated, so return the old status and update the current one accordingly
        old_status = authorization_request.status
        if old_status == AuthorizationRequestStatus.active and status in {
            AuthorizationRequestStatus.approved,
            AuthorizationRequestStatus.declined,
        }:
            authorization_request.status = AuthorizationRequestStatus.complete
        elif (
            old_status
            in {
                AuthorizationRequestStatus.approved,
                AuthorizationRequestStatus.declined,
            }
            and status == AuthorizationRequestStatus.active
        ):
            authorization_request.status = AuthorizationRequestStatus.complete

    return authorization_request, old_status

components.payment_gateway.subcomponents.authorizations.models.brokers.expense_category

ExpenseCategoryModelBroker

Bases: BaseModelBroker

SQL model broker for ExpenseCategory model.

See: https://www.notion.so/alaninsurance/SQL-Model-Brokers-90876b7e739c4195a6a2a4106139ed38?pvs=4 ⧉

create_expense_category classmethod 

create_expense_category(
    session,
    /,
    *,
    code,
    name,
    description=None,
    reference=None,
)

Create an Expense Category.

Parameters:

Name Type Description Default
code str

Machine-readable code of the category.

 required
name str

Human-readable name of the category.

 required
description str | None

Description of the category.

 None
reference str | None

Application-specific reference to the category.

 None

Returns:

Type Description
ExpenseCategory

The newly created Expense Category.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_category.py 
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
@classmethod
def create_expense_category(
    cls,
    session: Session,
    /,
    *,
    code: str,
    name: str,
    description: str | None = None,
    reference: str | None = None,
) -> ExpenseCategory:
    """Create an Expense Category.

    Args:
        code: Machine-readable code of the category.
        name: Human-readable name of the category.
        description: Description of the category.
        reference: Application-specific reference to the category.

    Returns:
        The newly created Expense Category.
    """
    expense_category = cls.model(
        code=code,
        name=name,
        description=description,
        reference=reference,
    )
    session.add(expense_category)
    session.flush()

    return expense_category

get_expense_category classmethod 

get_expense_category(session, /, id)

Get an Expense Category by its ID.

Parameters:

Name Type Description Default
id UUID

ID of the Expense Category.

 required

Returns:

Type Description
ExpenseCategory

The matched Expense Category.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_category.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@classmethod
def get_expense_category(
    cls,
    session: Session,
    /,
    id: UUID,
) -> ExpenseCategory:
    """Get an Expense Category by its ID.

    Args:
        id: ID of the Expense Category.

    Returns:
        The matched Expense Category.
    """
    expense_category: ExpenseCategory = session.execute(
        cls.select().filter(ExpenseCategory.id == id)
    ).scalar_one()
    return expense_category

get_expense_category_id_by_code classmethod 

get_expense_category_id_by_code(session, /, code)

Get Expense Category ID by its code.

Parameters:

Name Type Description Default
code str

Expense Category code.

 required

Returns:

Type Description
UUID

Expense Category ID.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_category.py 
 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
@classmethod
def get_expense_category_id_by_code(
    cls,
    session: Session,
    /,
    code: str,
) -> UUID:
    """Get Expense Category ID by its code.

    Args:
        code: Expense Category code.

    Returns:
        Expense Category ID.
    """
    expense_category: ExpenseCategory = session.execute(
        cls.select()
        .filter(
            and_(
                ExpenseCategory.code == code,
                ExpenseCategory.terminated_at.is_(None),
            )
        )
        .filter(
            and_(
                ExpenseCategory.code == code,
                ExpenseCategory.terminated_at.is_(None),
            )
        )
        .options(load_only(ExpenseCategory.id))
    ).scalar_one()
    return expense_category.id

list_expense_category_ids_by_codes classmethod 

list_expense_category_ids_by_codes(session, /, codes)

Get Expense Category IDs by codes.

Parameters:

Name Type Description Default
codes list[str]

List of Expense Category codes.

 required

Returns:

Type Description
list[UUID]

List of Expense Category IDs.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_category.py 
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
@classmethod
def list_expense_category_ids_by_codes(
    cls,
    session: Session,
    /,
    codes: list[str],
) -> list[UUID]:
    """Get Expense Category IDs by codes.

    Args:
        codes: List of Expense Category codes.

    Returns:
        List of Expense Category IDs.
    """
    expense_categories = session.execute(
        cls.select()
        .filter(
            and_(
                ExpenseCategory.code.in_(codes),
                ExpenseCategory.terminated_at.is_(None),
            )
        )
        .options(load_only(ExpenseCategory.id))
    ).scalars()
    return [expense_category.id for expense_category in expense_categories]

model class-attribute instance-attribute 

model = ExpenseCategory

terminate_expense_category classmethod 

terminate_expense_category(session, /, expense_category_id)

Terminate an Expense Category.

Parameters:

Name Type Description Default
expense_category_id UUID

ID of the Expense Category to terminate.

 required

Returns:

Type Description
ExpenseCategory

The terminated Expense Category.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_category.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@classmethod
def terminate_expense_category(
    cls,
    session: Session,
    /,
    expense_category_id: UUID,
) -> ExpenseCategory:
    """Terminate an Expense Category.

    Args:
        expense_category_id: ID of the Expense Category to terminate.

    Returns:
        The terminated Expense Category.
    """
    expense_category = cls.get_expense_category(session, expense_category_id)
    expense_category.terminated_at = datetime.now()

    return expense_category

components.payment_gateway.subcomponents.authorizations.models.brokers.expense_tracker

ExpenseTrackerModelBroker

Bases: BaseModelBroker

SQL model broker for ExpenseTracker model.

See: https://www.notion.so/alaninsurance/SQL-Model-Brokers-90876b7e739c4195a6a2a4106139ed38?pvs=4 ⧉

Note: all methods are designed to be fast and atomic to be used in a real-time context. This comes with tradeoffs. Any change must be carefully evaluated.

check_and_update_expensed_credits classmethod 

check_and_update_expensed_credits(
    session, /, lines_of_credit_ids, amount
)

Update expensed credits for all Lines of Credit by given amount if all have enough available credits, else do nothing. This method is atomic.

Parameters:

Name Type Description Default
lines_of_credit_ids list[UUID]

list of Line of Credit IDs to check and update

 required
amount int

amount to update expensed credits.

 required

Returns:

Type Description
bool

True if all Lines of Credit had enough available credits, else False (including if no matching rows have been found)
Note

This won't refresh the model instances, so you have to do it explicitly if for some reason you need the updated values after calling the method (which we don't in the case for which it was designed).

Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_tracker.py 
 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
@classmethod
def check_and_update_expensed_credits(
    cls,
    session: Session,
    /,
    lines_of_credit_ids: list[UUID],
    amount: int,
) -> bool:
    """Update expensed credits for all Lines of Credit by given amount if
    all have enough available credits, else do nothing. This method is
    atomic.

    Args:
        lines_of_credit_ids: list of Line of Credit IDs to check and update
        amount: amount to update expensed credits.

    Returns:
        True if all Lines of Credit had enough available credits, else False
            (including if no matching rows have been found)

    Note:
        This won't refresh the model instances, so you have to do it
        explicitly if for some reason you need the updated values after
        calling the method (which we don't in the case for which it was
        designed).
    """

    # CTE to check all rows meet condition
    check_all = (
        select(
            func.bool_and(
                ExpenseTracker.expensed_credits + amount
                <= ExpenseTracker.credits_limit
            ).label("all_valid"),
            func.count(ExpenseTracker.id).label("count"),
        )
        .where(ExpenseTracker.line_of_credit_id.in_(lines_of_credit_ids))
        .cte("check_all")
    )

    # Update with condition
    result = session.execute(
        ExpenseTracker.__table__.update()  # type: ignore[call-overload]
        .where(
            and_(
                ExpenseTracker.line_of_credit_id.in_(lines_of_credit_ids),
                exists()
                .where(check_all.c.all_valid)
                .where(check_all.c.count == len(lines_of_credit_ids)),
            )
        )
        .values(expensed_credits=ExpenseTracker.expensed_credits + amount)
        .returning(True)
    )

    return bool(result.first())

create_expense_tracker classmethod 

create_expense_tracker(
    session,
    /,
    *,
    line_of_credit_id,
    credits_limit=0,
    expensed_credits=0,
)

Create a new ExpenseTracker for a Line of Credit.

Parameters:

Name Type Description Default
line_of_credit_id UUID

Line of Credit ID.

 required
credits_limit int

Credits limit for the Line of Credit (in minor units).

 0
expensed_credits int

Expensed credits for the Line of Credit (in minor units).

 0

Returns:

Type Description
ExpenseTracker

The newly created Expense Tracker.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_tracker.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
51
52
53
54
55
@classmethod
def create_expense_tracker(
    cls,
    session: Session,
    /,
    *,
    line_of_credit_id: UUID,
    credits_limit: int = 0,
    expensed_credits: int = 0,
) -> ExpenseTracker:
    """Create a new ExpenseTracker for a Line of Credit.

    Args:
        line_of_credit_id: Line of Credit ID.
        credits_limit: Credits limit for the Line of Credit (in minor
            units).
        expensed_credits: Expensed credits for the Line of Credit (in
            minor units).

    Returns:
        The newly created Expense Tracker.
    """
    expense_tracker = ExpenseTracker(
        line_of_credit_id=line_of_credit_id,
        credits_limit=credits_limit,
        expensed_credits=expensed_credits,
    )
    session.add(expense_tracker)
    session.flush()

    return expense_tracker

get_expense_tracker_by_line_of_credit_id classmethod 

get_expense_tracker_by_line_of_credit_id(
    session, /, line_of_credit_id
)

Get an Expense Tracker by Line of Credit ID.

Parameters:

Name Type Description Default
line_of_credit_id UUID

Line of Credit ID.

 required

Returns:

Type Description
ExpenseTracker

The matched Expense Tracker.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_tracker.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@classmethod
def get_expense_tracker_by_line_of_credit_id(
    cls,
    session: Session,
    /,
    line_of_credit_id: UUID,
) -> ExpenseTracker:
    """Get an Expense Tracker by Line of Credit ID.

    Args:
        line_of_credit_id: Line of Credit ID.

    Returns:
        The matched Expense Tracker.
    """
    expense_tracker: ExpenseTracker = session.execute(
        cls.select().where(ExpenseTracker.line_of_credit_id == line_of_credit_id)
    ).scalar_one()
    return expense_tracker

model class-attribute instance-attribute 

model = ExpenseTracker

reset_credits classmethod 

reset_credits(
    session,
    /,
    *,
    line_of_credit_id,
    credits_limit,
    expensed_credits,
)

Reset the credits limit and expensed credits of a Line of Credit with provided values. This method is atomic.

Parameters:

Name Type Description Default
line_of_credit_id UUID

Line of Credit ID.

 required
credits_limit int

New credits limit (in minor units).

 required
expensed_credits int

New expensed credits (in minor units).

 required
Note

This won't refresh the model instances, so you have to do it explicitly if for some reason you need the updated values after calling the method (which we don't in the case for which it was designed).

Warning

This method doesn't check if the Line of Credit exists.

Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_tracker.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@classmethod
def reset_credits(
    cls,
    session: Session,
    /,
    *,
    line_of_credit_id: UUID,
    credits_limit: int,
    expensed_credits: int,
) -> None:
    """Reset the credits limit and expensed credits of a Line of Credit with
    provided values. This method is atomic.

    Args:
        line_of_credit_id: Line of Credit ID.
        credits_limit: New credits limit (in minor units).
        expensed_credits: New expensed credits (in minor units).

    Note:
        This won't refresh the model instances, so you have to do it
        explicitly if for some reason you need the updated values after
        calling the method (which we don't in the case for which it was
        designed).

    Warning:
        This method doesn't check if the Line of Credit exists.
    """

    session.execute(
        ExpenseTracker.__table__.update()
        .where(
            ExpenseTracker.line_of_credit_id == line_of_credit_id,
        )
        .values(
            credits_limit=credits_limit,
            expensed_credits=expensed_credits,
        )
    )

set_credits_limit classmethod 

set_credits_limit(
    session, /, *, line_of_credit_id, credits_limit
)

Set credit limits for a specific Line of Credit by given amount unconditionally. This method is atomic.

Parameters:

Name Type Description Default
line_of_credit_id UUID

Line of Credit ID to check and update

 required
credits_limit int

New credits limit (in minor units).

 required
Note

This won't refresh the model instances, so you have to do it explicitly if for some reason you need the updated values after calling the method (which we don't in the case for which it was designed).

Warning

This method doesn't check if the Lines of Credit exist.

Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_tracker.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
@classmethod
def set_credits_limit(
    cls,
    session: Session,
    /,
    *,
    line_of_credit_id: UUID,
    credits_limit: int,
) -> None:
    """Set credit limits for a specific Line of Credit by given amount
    unconditionally. This method is atomic.

    Args:
        line_of_credit_id: Line of Credit ID to check and update
        credits_limit: New credits limit (in minor units).

    Note:
        This won't refresh the model instances, so you have to do it
        explicitly if for some reason you need the updated values after
        calling the method (which we don't in the case for which it was
        designed).

    Warning:
        This method doesn't check if the Lines of Credit exist.
    """

    session.execute(
        ExpenseTracker.__table__.update()
        .where(
            ExpenseTracker.line_of_credit_id == line_of_credit_id,
        )
        .values(credits_limit=credits_limit)
    )

update_expensed_credits classmethod 

update_expensed_credits(
    session, /, lines_of_credit_ids, amount
)

Update expensed credits for all Lines of Credit by given amount unconditionally. This method is atomic.

Parameters:

Name Type Description Default
lines_of_credit_ids list[UUID]

list of Line of Credit IDs to check and update

 required
amount int

amount to update expensed credits.

 required
Note

This won't refresh the model instances, so you have to do it explicitly if for some reason you need the updated values after calling the method (which we don't in the case for which it was designed).

Warning

This method doesn't check if the Lines of Credit exist.

Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/expense_tracker.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@classmethod
def update_expensed_credits(
    cls,
    session: Session,
    /,
    lines_of_credit_ids: list[UUID],
    amount: int,
) -> None:
    """Update expensed credits for all Lines of Credit by given amount
    unconditionally. This method is atomic.

    Args:
        lines_of_credit_ids: list of Line of Credit IDs to check and update
        amount: amount to update expensed credits.

    Note:
        This won't refresh the model instances, so you have to do it
        explicitly if for some reason you need the updated values after
        calling the method (which we don't in the case for which it was
        designed).

    Warning:
        This method doesn't check if the Lines of Credit exist.
    """

    session.execute(
        ExpenseTracker.__table__.update()
        .where(
            ExpenseTracker.line_of_credit_id.in_(lines_of_credit_ids),
        )
        .values(expensed_credits=ExpenseTracker.expensed_credits + amount)
    )

components.payment_gateway.subcomponents.authorizations.models.brokers.line_of_credit

LineOfCreditModelBroker

Bases: BaseModelBroker

SQL model broker for LineOfCredit model.

See: https://www.notion.so/alaninsurance/SQL-Model-Brokers-90876b7e739c4195a6a2a4106139ed38?pvs=4 ⧉

autoload class-attribute instance-attribute 

autoload = {'expense_tracker': True}

create_line_of_credit classmethod 

create_line_of_credit(
    session,
    /,
    *,
    owner_type,
    owner_ref,
    expense_category_id,
)

Create a Line of Credit.

Parameters:

Name Type Description Default
owner_type str

Application-specific type of owner for the Line of Credit.

 required
owner_ref UUID

Application-specific reference to the owner of the Line of Credit.

 required
expense_category_id UUID

ID of the Expense Category the Line of Credit belongs to.

 required

Returns:

Type Description
LineOfCredit

The newly created Line of Credit.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/line_of_credit.py 
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
@classmethod
def create_line_of_credit(
    cls,
    session: Session,
    /,
    *,
    owner_type: str,
    owner_ref: UUID,
    expense_category_id: UUID,
) -> LineOfCredit:
    """Create a Line of Credit.

    Args:
        owner_type: Application-specific type of owner for the Line of Credit.
        owner_ref: Application-specific reference to the owner of the Line of Credit.
        expense_category_id: ID of the Expense Category the Line of Credit belongs to.

    Returns:
        The newly created Line of Credit.
    """
    line_of_credit = cls.model(
        owner_type=owner_type,
        owner_ref=owner_ref,
        expense_category_id=expense_category_id,
    )
    session.add(line_of_credit)
    session.flush()

    return line_of_credit

find_line_of_credit_id_for_owner_and_expense_category_id classmethod 

find_line_of_credit_id_for_owner_and_expense_category_id(
    session,
    /,
    *,
    owner_type,
    owner_ref,
    expense_category_id,
)

Get Line of Credit ID for a specific owner and Expense Category ID.

Parameters:

Name Type Description Default
owner_type str

Application-specific type of owner for the Line of Credit.

 required
owner_ref UUID

Application-specific reference to the owner of the Line of Credit.

 required
expense_category_id UUID

Expense Category IDs to resolve.

 required

Returns:

Type Description
UUID | None

Line of Credit ID or None if none found.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/line_of_credit.py 
 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
@classmethod
def find_line_of_credit_id_for_owner_and_expense_category_id(
    cls,
    session: Session,
    /,
    *,
    owner_type: str,
    owner_ref: UUID,
    expense_category_id: UUID,
) -> UUID | None:
    """Get Line of Credit ID for a specific owner and Expense Category ID.

    Args:
        owner_type: Application-specific type of owner for the Line of Credit.
        owner_ref: Application-specific reference to the owner of the Line of Credit.
        expense_category_id: Expense Category IDs to resolve.

    Returns:
        Line of Credit ID or None if none found.
    """
    line_of_credit = session.execute(
        cls.select(custom_autoload={})
        .filter(
            and_(
                LineOfCredit.owner_type == owner_type,
                LineOfCredit.owner_ref == owner_ref,
                LineOfCredit.expense_category_id == expense_category_id,
                LineOfCredit.terminated_at.is_(None),
            )
        )
        .options(load_only(LineOfCredit.id))
    ).scalar_one_or_none()
    return line_of_credit.id if line_of_credit else None

get_line_of_credit classmethod 

get_line_of_credit(session, /, id)

Get a Line of Credit by its ID.

Parameters:

Name Type Description Default
id UUID

ID of the Line of Credit.

 required

Returns:

Type Description
LineOfCredit

The matched Line of Credit.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/line_of_credit.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@classmethod
def get_line_of_credit(
    cls,
    session: Session,
    /,
    id: UUID,
) -> LineOfCredit:
    """Get a Line of Credit by its ID.

    Args:
        id: ID of the Line of Credit.

    Returns:
        The matched Line of Credit.
    """
    line_of_credit: LineOfCredit = session.execute(
        cls.select().filter(LineOfCredit.id == id)
    ).scalar_one()
    return line_of_credit

list_line_of_credit_ids_for_owner_and_expense_category_ids classmethod 

list_line_of_credit_ids_for_owner_and_expense_category_ids(
    session,
    /,
    *,
    owner_type,
    owner_ref,
    expense_category_ids,
)

Get Line of Credit IDs for a specific owner and a list of Expense Category IDs.

Parameters:

Name Type Description Default
owner_type str

Application-specific type of owner for the Line of Credit.

 required
owner_ref UUID

Application-specific reference to the owner of the Line of Credit.

 required
expense_category_ids list[UUID]

List of Expense Category IDs to resolve.

 required

Returns:

Type Description
list[UUID]

List of Line of Credit IDs.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/line_of_credit.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@classmethod
def list_line_of_credit_ids_for_owner_and_expense_category_ids(
    cls,
    session: Session,
    /,
    *,
    owner_type: str,
    owner_ref: UUID,
    expense_category_ids: list[UUID],
) -> list[UUID]:
    """Get Line of Credit IDs for a specific owner and a list of Expense Category IDs.

    Args:
        owner_type: Application-specific type of owner for the Line of Credit.
        owner_ref: Application-specific reference to the owner of the Line of Credit.
        expense_category_ids: List of Expense Category IDs to resolve.

    Returns:
        List of Line of Credit IDs.
    """
    lines_of_credit = (
        session.execute(
            cls.select(custom_autoload={})
            .filter(
                and_(
                    LineOfCredit.owner_type == owner_type,
                    LineOfCredit.owner_ref == owner_ref,
                    LineOfCredit.expense_category_id.in_(expense_category_ids),
                    LineOfCredit.terminated_at.is_(None),
                )
            )
            .options(load_only(LineOfCredit.id))
        )
        .scalars()
        .all()
    )
    return [line_of_credit.id for line_of_credit in lines_of_credit]

list_lines_of_credit_for_owner classmethod 

list_lines_of_credit_for_owner(
    session, /, *, owner_type, owner_ref
)

Get all Lines of Credit for a specific owner.

Parameters:

Name Type Description Default
owner_type str

Application-specific type of owner for the Line of Credit.

 required
owner_ref UUID

Application-specific reference to the owner of the Line of Credit.

 required

Returns:

Type Description
list[LineOfCredit]

List of Line of Credit IDs.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/line_of_credit.py 
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
@classmethod
def list_lines_of_credit_for_owner(
    cls,
    session: Session,
    /,
    *,
    owner_type: str,
    owner_ref: UUID,
) -> list[LineOfCredit]:
    """Get all Lines of Credit for a specific owner.

    Args:
        owner_type: Application-specific type of owner for the Line of Credit.
        owner_ref: Application-specific reference to the owner of the Line of Credit.

    Returns:
        List of Line of Credit IDs.
    """
    lines_of_credit = (
        session.execute(
            cls.select().filter(
                and_(
                    LineOfCredit.owner_type == owner_type,
                    LineOfCredit.owner_ref == owner_ref,
                    LineOfCredit.terminated_at.is_(None),
                )
            )
        )
        .scalars()
        .all()
    )
    return list(lines_of_credit)

model class-attribute instance-attribute 

model = LineOfCredit

terminate_line_of_credit classmethod 

terminate_line_of_credit(session, /, line_of_credit_id)

Terminate a Line of Credit.

Parameters:

Name Type Description Default
line_of_credit_id UUID

ID of the Line of Credit to terminate.

 required

Returns:

Type Description
LineOfCredit

The terminated Line of Credit.
Source code in components/payment_gateway/subcomponents/authorizations/models/brokers/line_of_credit.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@classmethod
def terminate_line_of_credit(
    cls,
    session: Session,
    /,
    line_of_credit_id: UUID,
) -> LineOfCredit:
    """Terminate a Line of Credit.

    Args:
        line_of_credit_id: ID of the Line of Credit to terminate.

    Returns:
        The terminated Line of Credit.
    """
    line_of_credit = cls.get_line_of_credit(session, line_of_credit_id)
    line_of_credit.terminated_at = datetime.now()

    return line_of_credit