Feature/tax remittance

mpesa_sdk/reversal/__init__.py

@@ -1,7 +1,6 @@
 from .schemas import (
     ReversalRequest,
     ReversalResponse,
-    ReversalReceiverIdentifierType,
     ReversalResultCallback,
     ReversalResultCallbackResponse,
     ReversalTimeoutCallback,
@@ -11,7 +10,6 @@
 
 __all__ = [
     "Reversal",
-    "ReversalReceiverIdentifierType",
     "ReversalRequest",
     "ReversalResponse",
     "ReversalResultCallback",

mpesa_sdk/reversal/schemas.py

@@ -8,28 +8,14 @@
 from typing import Optional, List
 
 
-class ReversalReceiverIdentifierType(str, Enum):
-    """Allowed values for ReceiverIdentifierType in Reversal requests."""
-
-    SHORT_CODE = "11"  # MSISDN
-
-
 class ReversalRequest(BaseModel):
     """Request schema for Transaction Reversal."""
 
     Initiator: str = Field(..., description="Username used to initiate the request.")
     SecurityCredential: str = Field(..., description="Encrypted security credential.")
-    CommandID: str = Field(
-        default="TransactionReversal", description="Type of transaction to perform."
-    )
     TransactionID: str = Field(..., description="Mpesa Transaction ID to reverse.")
     Amount: int = Field(..., description="Amount to reverse (in KES).")
     ReceiverParty: int = Field(..., description="Organization shortcode (6-9 digits).")
-    ReceiverIdentifierType: str = Field(
-        default=ReversalReceiverIdentifierType.SHORT_CODE.value,
-        description="Type of organization receiving the transaction.",
-        alias="RecieverIdentifierType",
-    )
     ResultURL: str = Field(..., description="URL for result notifications.")
     QueueTimeOutURL: str = Field(..., description="URL for timeout notifications.")
     Remarks: str = Field(
@@ -39,6 +25,9 @@ class ReversalRequest(BaseModel):
         None, description="Optional parameter (max 100 chars)."
     )
 
+    CommandID: str = "TransactionReversal"
+    RecieverIdentifierType: str = "11"
+
     model_config = ConfigDict(
         json_schema_extra={
             "example": {
@@ -48,7 +37,7 @@ class ReversalRequest(BaseModel):
                 "TransactionID": "LKXXXX1234",
                 "Amount": 100,
                 "ReceiverParty": 600610,
-                "ReceiverIdentifierType": "11",
+                "RecieverIdentifierType": "11",
                 "ResultURL": "https://ip:port/result",
                 "QueueTimeOutURL": "https://ip:port/timeout",
                 "Remarks": "Test",
@@ -60,22 +49,11 @@ class ReversalRequest(BaseModel):
     @model_validator(mode="before")
     @classmethod
     def validate(cls, values):
-        """Validate ReceiverIdentifierType and Remarks/Occasion length."""
-        cls._validate_identifier_type(values)
+        """Validates model."""
         cls._validate_remarks(values)
         cls._validate_occasion(values)
         return values
 
-    @classmethod
-    def _validate_identifier_type(cls, values):
-        identifier_type = values.get("ReceiverIdentifierType")
-        valid_types = [e.value for e in ReversalReceiverIdentifierType]
-        if identifier_type not in valid_types:
-            raise ValueError(
-                f"ReceiverIdentifierType must be one of {valid_types}, got '{identifier_type}'"
-            )
-        return values
-
     @classmethod
     def _validate_remarks(cls, values):
         remarks = values.get("Remarks")

mpesa_sdk/tax_remittance/__init__.py

@@ -0,0 +1,14 @@
+from .schemas import (
+    TaxRemittanceRequest,
+    TaxRemittanceResponse,
+    TaxRemittanceResultCallback,
+)
+from .tax_remittance import TaxRemittance
+
+__all__ = [
+    "TaxRemittance",
+    "TaxRemittanceRequest",
+    "TaxRemittanceResponse",
+    "TaxRemittanceResultCallback",
+    "TaxRemittanceResultCallbackResponse",
+]

mpesa_sdk/tax_remittance/schemas.py

@@ -0,0 +1,257 @@
+"""This module defines schemas for M-Pesa Tax Remittance API requests and responses."""
+
+from enum import Enum
+from pydantic import BaseModel, Field, ConfigDict
+from typing import Optional, List
+
+
+class TaxRemittanceRequest(BaseModel):
+    """Request schema for Tax Remittance."""
+
+    Initiator: str = Field(..., description="Username used to initiate the request.")
+    SecurityCredential: str = Field(..., description="Encrypted security credential.")
+    Amount: int = Field(..., description="Transaction amount (in KES).")
+    PartyA: int = Field(..., description="Shortcode from which money is deducted.")
+    AccountReference: str = Field(
+        ..., description="Payment registration number (PRN) issued by KRA."
+    )
+    Remarks: str = Field(
+        ..., description="Additional information for the transaction (max 100 chars)."
+    )
+    QueueTimeOutURL: str = Field(..., description="URL for timeout notifications.")
+    ResultURL: str = Field(..., description="URL for result notifications.")
+
+    PartyB: int = 572572
+    CommandID: str = "PayTaxToKRA"
+    SenderIdentifierType: int = 4
+    RecieverIdentifierType: int = 4
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "Initiator": "TaxPayer",
+                "SecurityCredential": "encrypted_credential",
+                "CommandID": "PayTaxToKRA",
+                "SenderIdentifierType": 4,
+                "ReceiverIdentifierType": 4,
+                "Amount": 239,
+                "PartyA": 888880,
+                "PartyB": 572572,
+                "AccountReference": "353353",
+                "Remarks": "OK",
+                "QueueTimeOutURL": "https://mydomain.com/b2b/remittax/queue/",
+                "ResultURL": "https://mydomain.com/b2b/remittax/result/",
+            }
+        }
+    )
+
+
+class TaxRemittanceResponse(BaseModel):
+    """Response schema for Tax Remittance request."""
+
+    OriginatorConversationID: Optional[str] = Field(
+        ..., description="Unique ID for the request message."
+    )
+    ConversationID: Optional[str] = Field(
+        ..., description="Unique ID for the transaction."
+    )
+    ResponseCode: str = Field(..., description="Status code, 0 means success.")
+    ResponseDescription: str = Field(..., description="Status message.")
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "OriginatorConversationID": "5118-111210482-1",
+                "ConversationID": "AG_20230420_2010759fd5662ef6d054",
+                "ResponseCode": "0",
+                "ResponseDescription": "Accept the service request successfully.",
+            }
+        }
+    )
+
+    def is_successful(self) -> bool:
+        """Check if the response indicates success."""
+        code = str(self.ResponseCode)
+        return code.strip("0") == "" and code != ""
+
+
+class TaxRemittanceResultParameter(BaseModel):
+    """Parameter item in Tax Remittance result notification."""
+
+    Key: str = Field(..., description="Parameter name.")
+    Value: str = Field(..., description="Parameter value.")
+
+
+class TaxRemittanceReferenceItem(BaseModel):
+    """Reference item in Tax Remittance result notification."""
+
+    Key: str = Field(..., description="Reference parameter name.")
+    Value: str = Field(..., description="Reference parameter value.")
+
+
+class TaxRemittanceReferenceData(BaseModel):
+    """Reference data in Tax Remittance result notification."""
+
+    ReferenceItem: List[TaxRemittanceReferenceItem] = Field(
+        ..., description="List of reference items."
+    )
+
+
+class TaxRemittanceResultParameters(BaseModel):
+    """Result parameters container."""
+
+    ResultParameter: List[TaxRemittanceResultParameter] = Field(
+        ..., description="List of result parameters."
+    )
+
+
+class TaxRemittanceResultMetadata(BaseModel):
+    """Metadata for Tax Remittance result notification."""
+
+    ResultType: int = Field(..., description="Type of result (0=Success, 1=Failure).")
+    ResultCode: int = Field(..., description="Result code (0=Success).")
+    ResultDesc: str = Field(..., description="Result description.")
+    OriginatorConversationID: str = Field(
+        ..., description="Originator conversation ID."
+    )
+    ConversationID: str = Field(..., description="Conversation ID.")
+    TransactionID: Optional[str] = Field(
+        None, description="M-Pesa transaction ID (if available)."
+    )
+    ResultParameters: Optional[TaxRemittanceResultParameters] = Field(
+        None, description="Result parameters container."
+    )
+    ReferenceData: Optional[TaxRemittanceReferenceData] = Field(
+        None, description="Reference data."
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "ResultType": 0,
+                "ResultCode": 0,
+                "ResultDesc": "The service request is processed successfully",
+                "OriginatorConversationID": "626f6ddf-ab37-4650-b882-b1de92ec9aa4",
+                "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b",
+                "TransactionID": "QKA81LK5CY",
+                "ResultParameters": {
+                    "ResultParameter": [
+                        {"Key": "Amount", "Value": "190.00"},
+                        {"Key": "Currency", "Value": "KES"},
+                        {"Key": "TransCompletedTime", "Value": "20221110110717"},
+                    ]
+                },
+                "ReferenceData": {
+                    "ReferenceItem": [
+                        {"Key": "BillReferenceNumber", "Value": "19008"},
+                        {
+                            "Key": "QueueTimeoutURL",
+                            "Value": "https://mydomain.com/b2b/remittax/queue/",
+                        },
+                    ]
+                },
+            }
+        }
+    )
+
+
+class TaxRemittanceResultCallback(BaseModel):
+    """Schema for Tax Remittance result notification sent to ResultURL."""
+
+    Result: TaxRemittanceResultMetadata = Field(..., description="Result metadata.")
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "Result": {
+                    "ResultType": 0,
+                    "ResultCode": 0,
+                    "ResultDesc": "The service request is processed successfully",
+                    "OriginatorConversationID": "626f6ddf-ab37-4650-b882-b1de92ec9aa4",
+                    "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b",
+                    "TransactionID": "QKA81LK5CY",
+                    "ResultParameters": {
+                        "ResultParameter": [
+                            {"Key": "Amount", "Value": "190.00"},
+                            {"Key": "Currency", "Value": "KES"},
+                            {"Key": "TransCompletedTime", "Value": "20221110110717"},
+                        ]
+                    },
+                    "ReferenceData": {
+                        "ReferenceItem": [
+                            {"Key": "BillReferenceNumber", "Value": "19008"},
+                            {
+                                "Key": "QueueTimeoutURL",
+                                "Value": "https://mydomain.com/b2b/remittax/queue/",
+                            },
+                        ]
+                    },
+                }
+            }
+        }
+    )
+
+    def is_successful(self) -> bool:
+        """Check if the result indicates success."""
+        return self.Result.ResultCode == 0
+
+
+class TaxRemittanceResultCallbackResponse(BaseModel):
+    """Schema for response sent back to Daraja API to acknowledge callback receipt."""
+
+    ResultCode: int = Field(default=0, description="Code indicating the result status.")
+    ResultDesc: str = Field(
+        default="Callback received successfully",
+        description="Description of the result.",
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "ResultCode": 0,
+                "ResultDesc": "Callback received successfully",
+            }
+        }
+    )
+
+
+class TaxRemittanceTimeoutCallback(BaseModel):
+    """Schema Tax Remittance sent to QueueTimeOutURL."""
+
+    Result: TaxRemittanceResultMetadata = Field(..., description="Result metadata.")
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "Result": {
+                    "ResultType": 1,
+                    "ResultCode": "1",
+                    "ResultDesc": "The service request timed out.",
+                    "OriginatorConversationID": "8521-4298025-1",
+                    "ConversationID": "AG_20181005_00004d7ee675c0c7ee0b",
+                }
+            }
+        }
+    )
+
+
+class TaxRemittanceTimeoutCallbackResponse(BaseModel):
+    """Schema for response to Tax Remittance timeout callback."""
+
+    ResultCode: int = Field(
+        default=0,
+        description="Result code (0=Success, other=Failure).",
+    )
+    ResultDesc: str = Field(
+        default="Timeout notification received and processed successfully.",
+        description="Result description.",
+    )
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "ResultCode": 0,
+                "ResultDesc": "Timeout notification received and processed successfully.",
+            }
+        }
+    )