Draft: Implement barcode payment, remove complexity and unnecessary code

santander_sdk/__init__.py

@@ -3,6 +3,7 @@
 from santander_sdk.api_client.helpers import (
     get_pix_key_type,
     document_type,
+    truncate_value,
 )
 
 from santander_sdk.pix import transfer_pix, get_transfer
@@ -48,4 +49,6 @@
     # Comom exceptions
     "SantanderRequestError",
     "SantanderError",
+    # Helpers
+    "truncate_value",
 ]

santander_sdk/api_client/helpers.py

@@ -1,3 +1,4 @@
+from datetime import date, datetime
 from decimal import ROUND_DOWN, Decimal
 import logging
 from itertools import cycle
@@ -158,3 +159,18 @@ def polling_until_condition(
             return result
         sleep(interval)
     raise TimeoutError("Timeout polling until condition is met")
+
+
+def today() -> date:
+    return datetime.now().date()
+
+
+def to_iso_date_string(value: str | date) -> str:
+    if isinstance(value, date):
+        return value.isoformat()
+    try:
+        return date.fromisoformat(value).isoformat()
+    except ValueError:
+        raise ValueError(
+            f"Invalid date format: {value}. Expected Iso Format YYYY-MM-DD."
+        )

santander_sdk/barcode_payment.py

@@ -0,0 +1,94 @@
+from datetime import date
+from decimal import Decimal as D
+from typing import Literal, cast
+
+from santander_sdk import SantanderApiClient
+from santander_sdk.api_client.exceptions import SantanderClientError
+from santander_sdk.api_client.helpers import (
+    to_iso_date_string,
+    truncate_value,
+)
+from santander_sdk.transfer_flow import SantanderPaymentFlow
+from santander_sdk.types import SantanderResponse, TransferResult
+
+SantanderBoletoResponse = SantanderResponse
+TransferBoletoReponse = TransferResult
+
+BASE_ENDPOINT = "/management_payments_partners/v1/workspaces/:workspaceid"
+BANKSLIP_ENDPOINT = f"{BASE_ENDPOINT}/bank_slip_payments"  # Boleto bancário
+BARCODE_ENDPOINT = f"{BASE_ENDPOINT}/barcode_payments"  # Boleto de arrecadação
+
+BarCodeType = Literal["barcode", "bankslip"]
+
+
+def pay_barcode(
+    client: SantanderApiClient,
+    barcode: str,
+    value: D,
+    payment_date: str | date,
+    tags: list[str] = [],
+    bar_code_type: BarCodeType = "bankslip",
+) -> TransferBoletoReponse:
+    payment_id = ""
+    flow = SantanderPaymentFlow(client, _resolve_endpoint(bar_code_type))
+    try:
+        # Step 1: Create the payment
+        create_barcode_response = flow.create_payment(
+            {
+                "code": barcode,
+                "tags": tags,
+                "paymentData": to_iso_date_string(payment_date),
+            }
+        )
+        payment_id = create_barcode_response.get("id")
+        if not payment_id:
+            raise SantanderClientError("Payment ID was not returned on creation")
+        if create_barcode_response.get("status") is None:
+            raise SantanderClientError("Payment status was not returned on creation")
+
+        # Step 2: Ensure the payment is ready to be confirmed
+        flow.ensure_ready_to_pay(create_barcode_response)
+
+        # Step 3: Confirm the payment
+        confirm_response = flow.confirm_payment(
+            {
+                "status": "AUTHORIZED",
+                "paymentValue": truncate_value(value),
+                "debitAccount": create_barcode_response.get("debitAccount"),
+                "finalPayer": create_barcode_response.get("finalPayer"),
+            },
+            payment_id,
+        )
+        return {
+            "success": True,
+            "request_id": flow.request_id,
+            "data": confirm_response,
+            "error": "",
+        }
+    except Exception as e:
+        error_message = str(e)
+        client.logger.error(error_message)
+        return {
+            "success": False,
+            "request_id": flow.request_id,
+            "error": error_message,
+            "data": None,
+        }
+
+
+def get_barcode_status(
+    client: SantanderApiClient, payment_id: str, barCodeType: BarCodeType
+) -> SantanderBoletoResponse:
+    if not payment_id:
+        raise ValueError("pix_payment_id not provided")
+    endpoint = _resolve_endpoint(barCodeType)
+    response = client.get(f"{endpoint}/{payment_id}")
+    return cast(SantanderBoletoResponse, response)
+
+
+def _resolve_endpoint(bar_code_type: BarCodeType) -> str:
+    if bar_code_type not in ("bankslip", "barcode"):
+        raise ValueError("bar_code_type must be 'bankslip' or 'barcode'")
+    if bar_code_type == "bankslip":
+        return BANKSLIP_ENDPOINT
+    return BARCODE_ENDPOINT

santander_sdk/pix.py

@@ -12,8 +12,8 @@
 from santander_sdk.transfer_flow import SantanderPaymentFlow
 from santander_sdk.types import (
     SantanderBeneficiary,
-    SantanderPixResponse,
-    TransferPixResult,
+    SantanderTransferResponse,
+    TransferResult,
 )
 
 PIX_ENDPOINT = "/management_payments_partners/v1/workspaces/:workspaceid/pix_payments"
@@ -26,7 +26,7 @@ def transfer_pix(
     description: str,
     tags: list[str] = [],
     id: uuid.UUID | str | None = None,
-) -> TransferPixResult:
+) -> TransferResult:
     transfer_flow = SantanderPaymentFlow(client, PIX_ENDPOINT)
 
     try:
@@ -37,7 +37,8 @@ def transfer_pix(
             pix_key, value, description, tags, id
         )
         create_pix_response = transfer_flow.create_payment(create_pix_dict)
-        if not create_pix_response.get("id"):
+        payment_id = create_pix_response.get("id")
+        if not payment_id:
             raise SantanderClientError("Payment ID was not returned on creation")
         if create_pix_response.get("status") is None:
             raise SantanderClientError("Payment status was not returned on creation")
@@ -47,9 +48,7 @@ def transfer_pix(
             "status": "AUTHORIZED",
             "paymentValue": truncate_value(value),
         }
-        confirm_response = transfer_flow.confirm_payment(
-            payment_data, create_pix_response.get("id")
-        )
+        confirm_response = transfer_flow.confirm_payment(payment_data, payment_id)
         return {
             "success": True,
             "request_id": transfer_flow.request_id,
@@ -69,11 +68,11 @@ def transfer_pix(
 
 def get_transfer(
     client: SantanderApiClient, pix_payment_id: str
-) -> SantanderPixResponse:
+) -> SantanderTransferResponse:
     if not pix_payment_id:
         raise ValueError("pix_payment_id not provided")
     response = client.get(f"{PIX_ENDPOINT}/{pix_payment_id}")
-    return cast(SantanderPixResponse, response)
+    return cast(SantanderTransferResponse, response)
 
 
 def _generate_create_pix_dict(
@@ -83,7 +82,7 @@ def _generate_create_pix_dict(
     tags: list = [],
     id: uuid.UUID | str | None = None,
 ) -> dict:
-    data = {
+    data: dict = {
         "tags": tags,
         "paymentValue": truncate_value(value),
         "remittanceInformation": description,

santander_sdk/transfer_flow.py

@@ -12,12 +12,7 @@
 from santander_sdk.api_client.helpers import (
     retry_one_time_on_request_exception,
 )
-from santander_sdk.types import (
-    ConfirmOrderStatus,
-    CreateOrderStatus,
-    OrderStatus,
-    SantanderPixResponse,
-)
+from santander_sdk.types import SantanderResponse
 
 MAX_UPDATE_STATUS_AFTER_CONFIRM = 120
 MAX_UPDATE_STATUS_BEFORE_CONFIRM = 10
@@ -36,35 +31,31 @@ def __init__(
         self.endpoint = endpoint
         self.request_id = None
 
-    def create_payment(self, data: dict) -> SantanderPixResponse:
-        response = cast(
-            SantanderPixResponse, self.client.post(self.endpoint, data=data)
-        )
+    def create_payment(self, data: dict) -> SantanderResponse:
+        response = cast(SantanderResponse, self.client.post(self.endpoint, data=data))
         self.request_id = response.get("id")
         self._check_for_rejected_error(response)
         self.client.logger.info("Payment created: ", response.get("id"))
         return response
 
     def ensure_ready_to_pay(self, confirm_data) -> None:
         payment_status = confirm_data.get("status")
-        if payment_status != CreateOrderStatus.READY_TO_PAY:
-            self.client.logger.info("PIX is not ready for payment", payment_status)
+        if payment_status != "READY_TO_PAY":
+            self.client.logger.info("Payment is not ready for payment", payment_status)
             self._payment_status_polling(
                 payment_id=confirm_data.get("id"),
-                until_status=[CreateOrderStatus.READY_TO_PAY],
+                until_status=["READY_TO_PAY"],
                 max_update_attemps=MAX_UPDATE_STATUS_BEFORE_CONFIRM,
             )
 
-    def confirm_payment(
-        self, confirm_data: dict, payment_id: str
-    ) -> SantanderPixResponse:
+    def confirm_payment(self, confirm_data: dict, payment_id: str) -> SantanderResponse:
         try:
             confirm_response = self._request_confirm_payment(confirm_data, payment_id)
         except SantanderRequestError as e:
             self.client.logger.error(str(e), payment_id, "checking current status")
             confirm_response = self._request_payment_status(payment_id)
 
-        if not confirm_response.get("status") == ConfirmOrderStatus.PAYED:
+        if not confirm_response.get("status") == "PAYED":
             try:
                 confirm_response = self._resolve_lazy_status_payed(
                     payment_id, confirm_response.get("status", "")
@@ -76,27 +67,27 @@ def confirm_payment(
         return confirm_response
 
     @retry_one_time_on_request_exception
-    def _request_payment_status(self, payment_id: str) -> SantanderPixResponse:
+    def _request_payment_status(self, payment_id: str) -> SantanderResponse:
         if not payment_id:
             raise ValueError("payment_id not provided")
         response = self.client.get(f"{self.endpoint}/{payment_id}")
-        response = cast(SantanderPixResponse, response)
+        response = cast(SantanderResponse, response)
         self._check_for_rejected_error(response)
         return response
 
     def _request_confirm_payment(
         self, confirm_data: dict, payment_id: str
-    ) -> SantanderPixResponse:
+    ) -> SantanderResponse:
         self.current_step = "CONFIRM"
         if not payment_id:
             raise ValueError("payment_id not provided")
         response = self.client.patch(f"{self.endpoint}/{payment_id}", data=confirm_data)
-        response = cast(SantanderPixResponse, response)
+        response = cast(SantanderResponse, response)
         self._check_for_rejected_error(response)
         return response
 
-    def _check_for_rejected_error(self, payment_response: SantanderPixResponse):
-        if not payment_response.get("status") == OrderStatus.REJECTED:
+    def _check_for_rejected_error(self, payment_response: SantanderResponse):
+        if not payment_response.get("status") == "REJECTED":
             return
         reject_reason = payment_response.get(
             "rejectReason", "Reason not returned by Santander"
@@ -106,20 +97,20 @@ def _check_for_rejected_error(self, payment_response: SantanderPixResponse):
         )
 
     def _resolve_lazy_status_payed(self, payment_id: str, current_status: str):
-        if not current_status == ConfirmOrderStatus.PENDING_CONFIRMATION:
+        if not current_status == "PENDING_CONFIRMATION":
             raise SantanderClientError(
                 f"Unexpected status after confirmation: {current_status}"
             )
         confirm_response = self._payment_status_polling(
             payment_id=payment_id,
-            until_status=[ConfirmOrderStatus.PAYED],
+            until_status=["PAYED"],
             max_update_attemps=MAX_UPDATE_STATUS_AFTER_CONFIRM,
         )
         return confirm_response
 
     def _payment_status_polling(
         self, payment_id: str, until_status: List[str], max_update_attemps: int
-    ) -> SantanderPixResponse:
+    ) -> SantanderResponse:
         response = None
 
         for attempt in range(1, max_update_attemps + 1):

santander_sdk/types.py

@@ -104,25 +104,9 @@ class SantanderDebitAccount(TypedDict):
     number: str
 
 
-class CreateOrderStatus:
-    READY_TO_PAY = "READY_TO_PAY"
-    PENDING_VALIDATION = "PENDING_VALIDATION"
-    REJECTED = "REJECTED"
-
-
-class ConfirmOrderStatus:
-    PAYED = "PAYED"
-    PENDING_CONFIRMATION = "PENDING_CONFIRMATION"
-    REJECTED = "REJECTED"
-
-
-class OrderStatus(ConfirmOrderStatus, CreateOrderStatus):
-    pass
-
-
-OrderStatusType = Literal[
-    "READY_TO_PAY", "PENDING_VALIDATION", "PAYED", "PENDING_CONFIRMATION", "REJECTED"
-]
+CreateOrderStatus = Literal["READY_TO_PAY", "PENDING_VALIDATION", "REJECTED"]
+ConfirmOrderStatus = Literal["PAYED", "PENDING_CONFIRMATION", "REJECTED"]
+OrderStatus = CreateOrderStatus | ConfirmOrderStatus
 
 
 class SantanderTransferResponse(TypedDict):
@@ -153,17 +137,17 @@ class SantanderTransferResponse(TypedDict):
     transaction: SantanderTransaction
     tags: list[str]
     paymentValue: str
-    status: OrderStatusType
+    status: OrderStatus
     dictCode: str | None
     dictCodeType: Literal["CPF", "CNPJ", "CELULAR", "EMAIL", "EVP"] | None
     beneficiary: SantanderBeneficiary | None
 
 
-SantanderPixResponse = SantanderTransferResponse | SantanderAPIErrorResponse
+SantanderResponse = SantanderTransferResponse | SantanderAPIErrorResponse
 
 
-class TransferPixResult(TypedDict):
+class TransferResult(TypedDict):
     success: bool
     request_id: str | None
-    data: SantanderPixResponse | None
+    data: SantanderResponse | None
     error: str

tests/conftest.py

@@ -1,3 +1,4 @@
+import re
 import pytest
 from santander_sdk.api_client.client import SantanderApiClient
 from santander_sdk.api_client.client_configuration import SantanderClientConfiguration
@@ -15,3 +16,19 @@ def client_instance():
             workspace_id=TEST_WORKSPACE_ID,
         )
     )
+
+
+@pytest.fixture
+def mock_auth(responses):
+    responses.add(
+        method=responses.POST,
+        url=re.compile(r".*oauth/v2/token"),
+        json={
+            "access_token": "mocked_access_token",
+            "token_type": "bearer",
+            "expires_in": 3600,
+            "scope": "read write",
+        },
+        status=200,
+    )
+    yield responses