Merge pull request #33 from RafaelJohn9/dev

Dev

Author: RafaelJohn9

mpesa_sdk/B2B_express_checkout/B2B_express_checkout.py

@@ -0,0 +1,51 @@
+"""B2BExpressCheckout: Handles M-Pesa B2B Express Checkout API interactions.
+
+This module provides functionality to initiate a B2B Express Checkout USSD Push transaction and handle result/timeout notifications
+using the M-Pesa API. Requires a valid access token for authentication and uses the HttpClient for HTTP requests.
+"""
+
+from pydantic import BaseModel, ConfigDict
+from mpesa_sdk.auth import TokenManager
+from mpesa_sdk.http_client import HttpClient
+
+from .schemas import (
+    B2BExpressCheckoutRequest,
+    B2BExpressCheckoutResponse,
+)
+
+
+class B2BExpressCheckout(BaseModel):
+    """Represents the B2B Express Checkout API client for M-Pesa operations.
+
+    https://developer.safaricom.co.ke/APIs/B2BExpressCheckout
+
+    Attributes:
+        http_client (HttpClient): HTTP client for making requests to the M-Pesa API.
+        token_manager (TokenManager): Manages access tokens for authentication.
+    """
+
+    http_client: HttpClient
+    token_manager: TokenManager
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def ussd_push(
+        self, request: B2BExpressCheckoutRequest
+    ) -> B2BExpressCheckoutResponse:
+        """Initiates a B2B Express Checkout USSD Push transaction.
+
+        Args:
+            request (B2BExpressCheckoutRequest): The B2B Express Checkout request data.
+
+        Returns:
+            B2BExpressCheckoutResponse: Response from the M-Pesa API.
+        """
+        url = "/v1/ussdpush/get-msisdn"
+        headers = {
+            "Authorization": f"Bearer {self.token_manager.get_token()}",
+            "Content-Type": "application/json",
+        }
+        response_data = self.http_client.post(
+            url, json=request.model_dump(mode="json"), headers=headers
+        )
+        return B2BExpressCheckoutResponse(**response_data)

mpesa_sdk/B2B_express_checkout/__init__.py

@@ -0,0 +1,16 @@
+from .schemas import (
+    B2BExpressCheckoutRequest,
+    B2BExpressCheckoutResponse,
+    B2BExpressCheckoutCallback,
+    B2BExpressCallbackResponse,
+)
+
+from .B2B_express_checkout import B2BExpressCheckout
+
+__all__ = [
+    "B2BExpressCheckout",
+    "B2BExpressCheckoutRequest",
+    "B2BExpressCheckoutResponse",
+    "B2BExpressCheckoutCallback",
+    "B2BExpressCallbackResponse",
+]

mpesa_sdk/B2B_express_checkout/schemas.py

@@ -0,0 +1,127 @@
+"""Schemas for M-PESA B2B Express Checkout APIs."""
+
+from pydantic import BaseModel, Field, ConfigDict, HttpUrl
+from typing import Optional
+
+
+class B2BExpressCheckoutRequest(BaseModel):
+    """Request schema for B2B Express Checkout USSD Push."""
+
+    primaryShortCode: int = Field(
+        ..., description="Merchant's till (debit party) shortcode/tillNumber."
+    )
+    receiverShortCode: int = Field(
+        ..., description="Vendor's paybill (credit party) shortcode."
+    )
+    amount: int = Field(..., description="Amount to be sent to vendor.")
+    paymentRef: str = Field(
+        ..., description="Reference for the payment (appears in text for merchant)."
+    )
+    callbackUrl: HttpUrl = Field(
+        ..., description="Vendor system endpoint for confirmation response."
+    )
+    partnerName: str = Field(..., description="Vendor's organization friendly name.")
+    RequestRefID: str = Field(..., description="Unique identifier for each request.")
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "primaryShortCode": 123456,
+                "receiverShortCode": 654321,
+                "amount": 100,
+                "paymentRef": "Invoice123",
+                "callbackUrl": "http://example.com/result",
+                "partnerName": "VendorName",
+                "RequestRefID": "550e8400-e29b-41d4-a716-446655440000",
+            }
+        }
+    )
+
+
+class B2BExpressCheckoutResponse(BaseModel):
+    """Acknowledgment response schema for B2B Express Checkout USSD Push."""
+
+    code: str = Field(
+        ..., description="Shows if the push was successful (0) or failed."
+    )
+    status: str = Field(..., description="USSD initiation status message.")
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "code": "0",
+                "status": "USSD Initiated Successfully",
+            }
+        }
+    )
+
+    def is_successful(self) -> bool:
+        """Check if the response indicates a successful USSD initiation."""
+        return self.code == "0"
+
+
+class B2BExpressCheckoutCallback(BaseModel):
+    """Callback response schema for B2B Express Checkout USSD Push."""
+
+    resultCode: str = Field(
+        ..., description="Status code: 0=success, other=fail/cancelled."
+    )
+    resultDesc: str = Field(..., description="Description of transaction result.")
+    amount: Optional[float] = Field(None, description="Amount initiated for payment.")
+    requestId: str = Field(..., description="Unique identifier of the request.")
+    paymentReference: Optional[str] = Field(
+        None, description="Reference for the payment."
+    )
+    resultType: Optional[str] = Field(
+        None, description="Status code for transaction sent to listener."
+    )
+    conversationID: Optional[str] = Field(
+        None, description="Global unique transaction request ID from M-Pesa."
+    )
+    transactionId: Optional[str] = Field(
+        None, description="Mpesa Receipt No of the transaction."
+    )
+    status: Optional[str] = Field(
+        None, description="Transaction status (SUCCESS/FAILED)."
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "resultCode": "0",
+                "resultDesc": "The service request is processed successfully.",
+                "amount": "71.0",
+                "requestId": "404e1aec-19e0-4ce3-973d-bd92e94c8021",
+                "resultType": "0",
+                "conversationID": "AG_20230426_2010434680d9f5a73766",
+                "transactionId": "RDQ01NFT1Q",
+                "status": "SUCCESS",
+            }
+        }
+    )
+
+    def is_successful(self) -> bool:
+        """Check if the callback indicates a successful transaction."""
+        return self.resultCode == "0"
+
+
+class B2BExpressCallbackResponse(BaseModel):
+    """Response schema for B2B Express Checkout callback."""
+
+    ResultCode: int = Field(
+        default=0,
+        description="Result code (0=Success, other=Failure).",
+    )
+    ResultDesc: str = Field(
+        default="Callback received successfully.",
+        description="Result description.",
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "ResultCode": 0,
+                "ResultDesc": "Callback received successfully.",
+            }
+        }
+    )

mpesa_sdk/B2C/B2C.py

@@ -0,0 +1,47 @@
+"""B2C: Handles M-Pesa B2C (Business to Customer) API interactions.
+
+This module provides functionality to initiate B2C payments using the M-Pesa API.
+Requires a valid access token for authentication and uses the HttpClient for HTTP requests.
+"""
+
+from pydantic import BaseModel, ConfigDict
+from mpesa_sdk.auth import TokenManager
+from mpesa_sdk.http_client import HttpClient
+
+from .schemas import (
+    B2CRequest,
+    B2CResponse,
+)
+
+
+class B2C(BaseModel):
+    """Represents the B2C API client for M-Pesa Business to Customer operations.
+
+    https://developer.safaricom.co.ke/APIs/BusinessToCustomerPayment
+
+    Attributes:
+        http_client (HttpClient): HTTP client for making requests to the M-Pesa API.
+        token_manager (TokenManager): Manages access tokens for authentication.
+    """
+
+    http_client: HttpClient
+    token_manager: TokenManager
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def send_payment(self, request: B2CRequest) -> B2CResponse:
+        """Initiates a B2C payment request.
+
+        Args:
+            request (B2CRequest): The payment request details.
+
+        Returns:
+            B2CResponse: Response from the M-Pesa API after payment initiation.
+        """
+        url = "/mpesa/b2c/v3/paymentrequest"
+        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 B2CResponse(**response_data)

mpesa_sdk/B2C/__init__.py

@@ -0,0 +1,24 @@
+from .schemas import (
+    B2CCommandIDType,
+    B2CRequest,
+    B2CResponse,
+    B2CResultParameter,
+    B2CResultMetadata,
+    B2CResultCallback,
+    B2CTimeoutCallbackResponse,
+    B2CTimeoutCallback,
+    B2CTimeoutCallbackResponse,
+)
+from .B2C import B2C
+
+__all__ = [
+    "B2C",
+    "B2CCommandIDType",
+    "B2CRequest",
+    "B2CResponse",
+    "B2CResultParameter",
+    "B2CResultMetadata",
+    "B2CResultCallback",
+    "B2CTimeoutCallbackResponse",
+    "B2CTimeoutCallback",
+]