Skip to content

Model brokers

components.payment_gateway.subcomponents.merchants.models.brokers.merchant_info

MerchantInfoModelBroker

Bases: BaseModelBroker

find_merchant_info_by_merchant_id classmethod 

find_merchant_info_by_merchant_id(session, /, merchant_id)
Source code in components/payment_gateway/subcomponents/merchants/models/brokers/merchant_info.py 
37
38
39
40
41
42
43
44
45
46
47
@classmethod
def find_merchant_info_by_merchant_id(
    cls,
    session: Session,
    /,
    merchant_id: str,
) -> MerchantInfo | None:
    merchant_info: MerchantInfo | None = session.execute(
        cls.select().filter(cls.model.merchant_id == merchant_id.strip())
    ).scalar_one_or_none()
    return merchant_info

get_merchant_info classmethod 

get_merchant_info(session, /, id)
Source code in components/payment_gateway/subcomponents/merchants/models/brokers/merchant_info.py 
25
26
27
28
29
30
31
32
33
34
35
@classmethod
def get_merchant_info(
    cls,
    session: Session,
    /,
    id: UUID,
) -> MerchantInfo:
    merchant_info: MerchantInfo = session.execute(
        cls.select().filter(cls.model.id == id)
    ).scalar_one()
    return merchant_info

model class-attribute instance-attribute 

model = MerchantInfo

upsert_merchant_info classmethod 

upsert_merchant_info(session, /, *, merchant_id, **data)
Source code in components/payment_gateway/subcomponents/merchants/models/brokers/merchant_info.py 
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 upsert_merchant_info(
    cls,
    session: Session,
    /,
    *,
    merchant_id: str,
    **data: Unpack[MerchantInfoUpdate],
) -> MerchantInfo:
    insert_stmt = insert(cls.model).values(
        merchant_id=merchant_id,
        **data,
    )
    conflict_set = {
        "merchant_id": insert_stmt.excluded["merchant_id"],
    }
    for key in data.keys():
        conflict_set[key] = insert_stmt.excluded[key]
    upsert_stmt = insert_stmt.on_conflict_do_update(
        index_elements=["merchant_id"],  # 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)

MerchantInfoUpdate

Bases: TypedDict

acquirer_id instance-attribute 

acquirer_id

city instance-attribute 

city

country instance-attribute 

country

mcc instance-attribute 

mcc

name instance-attribute 

name

postal_code instance-attribute 

postal_code