Merge pull request #9 from RafaelJohn9/feature/qrcode-api

Feature/qrcode api

Author: RafaelJohn9

mpesa_sdk/dynamic_qr_code/__init__.py

@@ -0,0 +1,13 @@
+from .dynamic_qr_code import DynamicQRCode
+from .schemas import (
+    DynamicQRGenerateRequest,
+    DynamicQRGenerateResponse,
+    DynamicQRTransactionType,
+)
+
+__all__ = [
+    "DynamicQRCode",
+    "DynamicQRGenerateRequest",
+    "DynamicQRGenerateResponse",
+    "DynamicQRTransactionType",
+]

mpesa_sdk/dynamic_qr_code/dynamic_qr_code.py

@@ -0,0 +1,49 @@
+"""Dynamic QR Code: Generates a dynamic M-Pesa QR Code.
+
+This module provides functionality to generate a Dynamic QR code using the M-Pesa API.
+It requires a valid access token for authentication and uses the HttpClient for making HTTP requests.
+"""
+
+from pydantic import BaseModel, ConfigDict
+from mpesa_sdk.auth import TokenManager
+from mpesa_sdk.http_client import HttpClient
+
+from .schemas import (
+    DynamicQRGenerateRequest,
+    DynamicQRGenerateResponse,
+)
+
+
+class DynamicQRCode(BaseModel):
+    """Represents the request payload for generating a Dynamic M-Pesa QR code.
+
+    https://developer.safaricom.co.ke/APIs/DynamicQR
+
+    Attributes:
+        http_client (HttpClient): The HTTP client used to make requests to the M-Pesa API.
+        token_manager (TokenManager): The token manager for handling access tokens.
+    """
+
+    http_client: HttpClient
+    token_manager: TokenManager
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def generate(self, request: DynamicQRGenerateRequest) -> DynamicQRGenerateResponse:
+        """Generates a Dynamic M-Pesa QR Code.
+
+        Args:
+            request (DynamicQRGenerateRequest): The request data for generating the QR code.
+
+        Returns:
+            DynamicQRGenerateResponse: The response from the M-Pesa API after generating the QR code.
+        """
+        url = "/mpesa/qrcode/v1/generate"
+        headers = {
+            "Authorization": f"Bearer {self.token_manager.get_token()}",
+            "Content-Type": "application/json",
+        }
+
+        response_data = self.http_client.post(url, json=dict(request), headers=headers)
+
+        return DynamicQRGenerateResponse(**response_data)

mpesa_sdk/dynamic_qr_code/schemas.py

@@ -0,0 +1,167 @@
+"""mpesa_sdk.dynamic_qr_code.schemas.
+
+This module defines the schemas for generating a dynamic M-Pesa QR code.
+It includes request and response models using Pydantic for validation and serialization.
+"""
+
+from enum import Enum
+from typing import Optional
+from pydantic import BaseModel, Field, ConfigDict
+from pydantic import model_validator
+from mpesa_sdk.utils.phone import normalize_phone_number
+
+
+class DynamicQRTransactionType(str, Enum):
+    """Enum representing the supported transaction types for Dynamic QR."""
+
+    BUY_GOODS = "BG"  # Pay Merchant (Buy Goods)
+    WITHDRAW_CASH = "WA"  # Withdraw Cash at Agent Till
+    PAYBILL = "PB"  # Paybill or Business number
+    SEND_MONEY = "SM"  # Send Money (Mobile number)
+    SEND_TO_BUSINESS = "SB"  # Sent to Business (Bu.siness number CPI in MSISDN format)
+
+
+class DynamicQRGenerateRequest(BaseModel):
+    """Represents the request payload for generating a Dynamic QR code.
+
+    https://developer.safaricom.co.ke/APIs/DynamicQRCode
+
+    Attributes:
+        MerchantName (str): Name of the Company/M-Pesa Merchant Name.
+        RefNo (str): Transaction Reference. For Paybill, Withdraw Cash, and similar transactions, this is where you enter your account number.
+        Amount (float): The total amount for the sale/transaction.
+        TrxCode (DynamicQRTransactionType): Transaction Type.
+        CPI (str): Credit Party Identifier (Mobile Number, Business Number, Agent Till, Paybill, etc.).
+        Size (str): Size of the QR code image in pixels (always a square image).
+    """
+
+    MerchantName: str = Field(
+        ...,
+        description="Name of the Company/M-Pesa Merchant Name.",
+        examples=["TEST SUPERMARKET"],
+    )
+    RefNo: str = Field(
+        ...,
+        description="Transaction Reference. For Paybill, Withdraw Cash, and similar transactions, this is where you enter your account number.",
+        examples=["Invoice Test", "xewr34fer4t", "ACC12345"],
+    )
+    Amount: int = Field(
+        ...,
+        description="The total amount for the sale/transaction.",
+        examples=[1, 2000],
+        gt=0,
+    )
+    TrxCode: str = Field(
+        ...,
+        description="Transaction Type. Supported: BG, WA, PB, SM, SB.",
+        examples=["BG"],
+    )
+
+    CPI: str = Field(
+        ...,
+        description="Credit Party Identifier. Can be a Mobile Number, Business Number, Agent Till, Paybill or Business number, or Merchant Buy Goods.",
+        examples=["373132", "174379"],
+    )
+    Size: str = Field(
+        ...,
+        description="Size of the QR code image in pixels. QR code image will always be a square image.",
+        examples=["300"],
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "MerchantName": "TEST SUPERMARKET",
+                "RefNo": "Invoice Test",
+                "Amount": 1,
+                "TrxCode": "BG",
+                "CPI": "373132",
+                "Size": "300",
+            }
+        }
+    )
+
+    @classmethod
+    def validate_trx_code(cls, value):
+        """Validates the transaction code against the DynamicQRTransactionType enum."""
+        try:
+            DynamicQRTransactionType(value)
+        except ValueError:
+            raise ValueError(
+                f"TrxCode must be one of: {[e.value for e in DynamicQRTransactionType]}"
+            )
+        return value
+
+    @classmethod
+    def normalize_cpi_for_send_money(cls, values):
+        """If TrxCode is SEND_MONEY, normalize the CPI (mobile number).
+
+        - If it starts with '0', replace with '254'
+        - If it starts with '+254', replace with '254'
+        - If it can't be normalized, raise ValueError
+        """
+        trx_code = values.get("TrxCode")
+        cpi = values.get("CPI")
+        if trx_code == DynamicQRTransactionType.SEND_MONEY.value and isinstance(
+            cpi, str
+        ):
+            normalized = normalize_phone_number(cpi)
+            if normalized is None:
+                raise ValueError(
+                    "CPI for SEND_MONEY must be a valid Kenyan phone number starting with '0', '+254', or '254'."
+                )
+            values["CPI"] = normalized
+        return values
+
+    @model_validator(mode="before")
+    def validate(cls, values):
+        """Validates the TrxCode field before model validation."""
+        # Validate the TrxCode field
+        trx_code = values.get("TrxCode")
+        if trx_code is not None:
+            cls.validate_trx_code(trx_code)
+
+        # Normalize CPI for SEND_MONEY transaction type
+        values = cls.normalize_cpi_for_send_money(values)
+
+        return values
+
+
+class DynamicQRGenerateResponse(BaseModel):
+    """Represents the response returned after generating a Dynamic QR code.
+
+    https://developer.safaricom.co.ke/APIs/DynamicQRCode
+
+    Attributes:
+        ResponseCode (str): Used to return the Transaction Type (alpha-numeric string).
+        RequestID (str): Unique identifier for the request.
+        ResponseDescription (str): Description of the transaction status.
+        QRCode (str): QR Code Image/Data/String (base64 or similar).
+    """
+
+    ResponseCode: str = Field(
+        ...,
+        description="Used to show if the transaction was successful or not. 00 indicates success.",
+        examples=["00"],
+    )
+    ResponseDescription: str = Field(
+        ...,
+        description="This is a response describing the status of the transaction.",
+        examples=["QR Code Successfully Generated."],
+    )
+    QRCode: str = Field(
+        ...,
+        description="QR Code Image/Data/String.",
+        examples=["iVBORw0KGgoAAAANSUhEUgAAASwAAAEsCAIAAAD2HxkiAAAHtElEQVR42..."],
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "ResponseCode": "00",
+                "RequestID": "16738-27456357-1",
+                "ResponseDescription": "QR Code Successfully Generated.",
+                "QRCode": "iVBORw0KGgoAAAANSUhEUgAAASwAAAEsCAIAAAD2HxkiAAAHtElEQVR42...",
+            }
+        }
+    )

mpesa_sdk/utils/phone.py

@@ -0,0 +1,38 @@
+"""Utility functions for handling phone numbers in the M-Pesa SDK."""
+
+
+def normalize_phone_number(phone: str) -> str | None:
+    """Normalize a Kenyan phone number to the '2547XXXXXXXX' format.
+
+    - If it starts with '0', replace with '254'
+    - If it starts with '+254', replace with '254'
+    - If it starts with '254', return as is
+    - Handles whitespace anywhere in the number
+    - Otherwise, return None
+
+    Args:
+        phone (str): The phone number to normalize.
+
+    Returns:
+        str | None: Normalized phone number or None if invalid.
+    """
+    if not isinstance(phone, str):
+        return None
+    phone = phone.strip().replace(" ", "")
+    normalized = None
+    if phone.startswith("+254") and len(phone) == 13:
+        normalized = "254" + phone[4:]
+    elif phone.startswith("0") and len(phone) == 10:
+        normalized = "254" + phone[1:]
+    elif phone.startswith("254") and len(phone) == 12:
+        normalized = phone
+
+    if (
+        normalized
+        and normalized.isdigit()
+        and len(normalized) == 12
+        and normalized.startswith("2547")
+    ):
+        return normalized
+    else:
+        return None

tests/integration/dynamic_qr_code/test_dynamic_qr_code_e2e.py

@@ -0,0 +1,46 @@
+"""End-to-End Test for M-Pesa Dynamic QR Code Generation."""
+
+import os
+import pytest
+from dotenv import load_dotenv
+
+from mpesa_sdk.dynamic_qr_code import (
+    DynamicQRGenerateRequest,
+    DynamicQRCode,
+    DynamicQRTransactionType,
+)
+from mpesa_sdk.auth import TokenManager
+from mpesa_sdk.http_client.mpesa_http_client import MpesaHttpClient
+
+pytestmark = pytest.mark.live
+
+load_dotenv()
+
+
+@pytest.fixture
+def dynamic_qr_service():
+    """Initialize the M-Pesa Dynamic QR Code service with authentication."""
+    http_client = MpesaHttpClient(env=os.getenv("MPESA_ENV", "sandbox"))
+    token_manager = TokenManager(
+        http_client=http_client,
+        consumer_key=os.getenv("MPESA_CONSUMER_KEY"),
+        consumer_secret=os.getenv("MPESA_CONSUMER_SECRET"),
+    )
+    return DynamicQRCode(http_client=http_client, token_manager=token_manager)
+
+
+def test_dynamic_qr_code_generate(dynamic_qr_service):
+    """End-to-end test for M-Pesa Dynamic QR Code generation."""
+    request = DynamicQRGenerateRequest(
+        MerchantName="Test Supermarket",
+        RefNo="xewr34fer4t",
+        Amount=200,
+        TrxCode=DynamicQRTransactionType.BUY_GOODS,
+        CPI="373132",
+        Size="300",
+    )
+    response = dynamic_qr_service.generate(request)
+    # Basic assertions - adapt as needed for your SDK's response structure
+    assert response is not None
+    assert hasattr(response, "QRCode") or hasattr(response, "qr_code")
+    assert getattr(response, "QRCode", None) or getattr(response, "qr_code", None)

tests/integration/mpesa_express/test_stk_push_e2e.py

@@ -113,7 +113,7 @@ def test_stk_push_full_e2e_with_query(stk_service, fastapi_server, ngrok_tunnel)
     callback = None
     for _ in range(30):
         time.sleep(1)
-        r = requests.get(f"{callback_base_url}/latest")
+        r = requests.get(f"{callback_base_url}/latest", timeout=45)
         if r.status_code == 200:
             callback_received = True
             callback_json = r.json()["parsed"]