Pydantic models/schema for Daraja M-Pesa Express API.

mpesa-schema.py

# Python 3.9.6+

from typing import Literal, Optional, Union

import humps
from pydantic import BaseModel, Field, validator
from typing_extensions import Annotated

### Queue charge ###


# TODO: Create custom timestamp de/serializer
class PushRequest(BaseModel):
    business_short_code: int  # 654321
    password: str  # base64(shortcode,passkey,timestamp)
    timestamp: str  # yyyymmddhhmmss
    transaction_type: Literal["CustomerPayBillOnline"]
    amount: int  #
    party_a: int  # 2547XXXXXXXX
    party_b: int  # Same as shortcode
    phone_number: int  # Same as party_a
    call_back_url: str = Field(..., alias="CallBackURL")
    account_reference: str = Field(..., max_length=12)  # Max 12-chars
    transaction_desc: Optional[str] = Field(None, max_length=13)  # Max 13-chars

    @validator("account_reference")
    def account_reference_alphanumeric(cls, v: str):
        assert v.isalnum(), "must be alphanumeric"
        return v

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


# Status 500, On payment queue error
class PushQueueError(BaseModel):
    request_id: Optional[str] = Field(None, alias="requestID") # 168-...-1
    error_code: Optional[str] = None  # ws..., Conflicting docs?
    error_message: Optional[str] = None  # You can display this message to the customer

    class Config:
        alias_generator = humps.camelize  # type: ignore[reportPrivateImportUsage]


class PushQueueSuccess(BaseModel):   
    merchant_request_id: str = Field(..., alias="MerchantRequestID") # 168-...-1
    checkout_request_id: str = Field(..., alias="CheckoutRequestID") # ws..., Conflicting docs?
    response_description: str  # "The service request has been accepted successfully", "The service request has failed"
    response_code: str  # "0" is success
    customer_message: str  # Display to customer

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]

## Callback ###


class AmountItem(BaseModel):
    name: Literal["Amount"]
    value: float

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class ReceiptNumberItem(BaseModel):
    name: Literal["MpesaReceiptNumber"]
    value: str

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class TransactionDateItem(BaseModel):
    name: Literal["TransactionDate"]
    value: int

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class PhoneNumberItem(BaseModel):
    name: Literal["PhoneNumber"]
    value: int  # Send only int amounts

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class BalanceItem(BaseModel):
    name: Literal["Balance"]
    value: Optional[int] = None

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


Item = Annotated[
    Union[
        AmountItem, ReceiptNumberItem, TransactionDateItem, PhoneNumberItem, BalanceItem
    ],
    Field(discriminator="name"),
]

Items = list[Item]


class CallbackMetada(BaseModel):
    # !Are the items always ordered?
    item: Items

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class CallbackBase(BaseModel):
    merchant_request_id: str = Field(..., alias="MerchantRequestID")  # 168-...-1
    checkout_request_id: str = Field(..., alias="CheckoutRequestID")  # 168-...-1
    # result_code: int  # 0 for success
    result_desc: str

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class StkCallbackSuccess(CallbackBase):
    callback_metadata: CallbackMetada
    result_code: Literal[0]

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class StkCallbackError(CallbackBase):
    result_code: int

    class Config:
        # Correctly discriminate with StkCallbackSuccess
        # Since this object has fewer fields
        extra = "forbid"


class CallbackResponseBody(BaseModel):
    stk_callback: Union[StkCallbackSuccess, StkCallbackError] = Field(
        ...,
        alias="stkCallback",
    )


# On successful payment
class CallbackSuccess(BaseModel):
    body: CallbackResponseBody

    class Config:
        alias_generator = humps.pascalize  # type: ignore[reportPrivateImportUsage]


class CallbackError(BaseModel):
    request_id: str
    error_code: str
    error_message: str
      
    class Config:
        alias_generator = humps.camelize  # type: ignore[reportPrivateImportUsage]


CallbackResponse = Union[CallbackSuccess, CallbackError]