wip payments (iugu)

orders-events/app/events/billing/cancel_enrollment.py

@@ -82,7 +82,7 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
     except Exception as exc:
         logger.exception(
             exc,
-            keypair={'pk': pk, 'sk': sk},
+            keypair={'id': pk, 'sk': sk},
         )
         return False
     else:

orders-events/app/events/payments/append_fee.py

@@ -0,0 +1,20 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import (
+    EventBridgeEvent,
+    event_source,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from layercake.dynamodb import (
+    DynamoDBPersistenceLayer,
+)
+
+from boto3clients import dynamodb_client
+from config import ORDER_TABLE
+
+logger = Logger(__name__)
+dyn = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
+
+
+@event_source(data_class=EventBridgeEvent)
+@logger.inject_lambda_context
+def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool: ...

orders-events/app/events/payments/charge_credit_card.py

@@ -0,0 +1,20 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import (
+    EventBridgeEvent,
+    event_source,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from layercake.dynamodb import (
+    DynamoDBPersistenceLayer,
+)
+
+from boto3clients import dynamodb_client
+from config import ORDER_TABLE
+
+logger = Logger(__name__)
+dyn = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
+
+
+@event_source(data_class=EventBridgeEvent)
+@logger.inject_lambda_context
+def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool: ...

orders-events/app/events/payments/create_invoice.py

@@ -0,0 +1,30 @@
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import (
+    EventBridgeEvent,
+    event_source,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+from layercake.dynamodb import (
+    DynamoDBPersistenceLayer,
+)
+
+from boto3clients import dynamodb_client
+from config import ORDER_TABLE
+
+logger = Logger(__name__)
+dyn = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
+
+
+@event_source(data_class=EventBridgeEvent)
+@logger.inject_lambda_context
+def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
+    new_image = event.detail['new_image']
+    order_id = new_image['id']
+
+    doc = {
+        'id': order_id,
+        'sk': 'IUGU',
+        'invoice_id': '',
+    }
+
+    return True

orders-events/app/iugu.py

@@ -0,0 +1,247 @@
+"""
+Notes:
+-----
+- `%d` Day of the month as a zero-padded decimal number. Ex: 01, 02, …, 31
+- `%m` Month as a zero-padded decimal number. Ex: 01, 02, …, 12
+- `%Y` Year with century as a decimal number. Ex: 0001, 0002, …, 2013, 2014
+
+- https://docs.python.org/3/library/datetime.html#strftime-and-strptime-format-codes
+
+Documentation:
+--------------
+- https://dev.iugu.com/reference/criar-fatura
+- https://dev.iugu.com/reference/criar-token
+- https://dev.iugu.com/reference/cobranca-direta
+- https://support.iugu.com/hc/pt-br/articles/212456346-Usar-cart%C3%B5es-de-teste-em-modo-de-teste
+"""
+
+from dataclasses import dataclass
+from enum import Enum
+from urllib.parse import ParseResult, urlparse
+
+import requests
+from aws_lambda_powertools import Logger
+from pydantic import BaseModel, HttpUrl
+
+# from data_classes.invoice import Invoice, Pix
+# from data_classes.order import CreditCard, Order, PaymentMethod, Status
+
+logger = Logger(__name__)
+
+
+class Status(Enum):
+    PAID = 'PAID'
+    DECLINED = 'DECLINED'
+
+
+class PaymentMethod(Enum):
+    PIX = 'PIX'
+    BANK_SLIP = 'BANK_SLIP'
+    CREDIT_CARD = 'CREDIT_CARD'
+
+
+class Order: ...
+
+
+class CreditCard: ...
+
+
+class BankSlip(BaseModel):
+    digitable_line: str
+    bank_slip_url: HttpUrl
+
+
+@dataclass
+class Credentials:
+    account_id: str
+    api_token: str
+    test_mode: bool = True
+
+
+@dataclass
+class Token(str):
+    id: str
+
+
+@dataclass
+class Transaction:
+    status: Status
+    response: dict
+
+
+class Iugu:
+    base_url: ParseResult = urlparse('https://api.iugu.com')
+
+    def __init__(self, credentials: Credentials) -> None:
+        self.credentials = credentials
+
+    def url(self, **kwargs) -> str:
+        return self.base_url._replace(
+            query=f'api_token={self.credentials.api_token}', **kwargs
+        ).geturl()
+
+    def create_invoice(
+        self,
+        order: Order,
+        postback_url: ParseResult | str,
+    ) -> Invoice:
+        """
+        O que é uma fatura?
+        -------------------
+        A fatura é uma forma de cobrança da iugu que possibilitar efetuar cobranças
+        com os métodos cartão de crédito, boleto e PIX, fora isso nela é possível
+        definir como será realizado o split de pagamentos.
+
+        Por que usar a chamada de fatura?
+        ---------------------------------
+        No response dessa chamada é retornado uma url na propriedade secure_url,
+        onde o cliente final pode acessar e efetuar o pagamento em um checkout da
+        IUGU.
+        """
+
+        url = self.url(path='/v1/invoices')
+
+        if isinstance(postback_url, str):
+            postback_url = urlparse(postback_url)
+
+        items = [
+            {
+                'description': item.name,
+                'price_cents': int(item.unit_price * 100),
+                'quantity': item.quantity,
+            }
+            for item in order.items
+        ]
+        payload = {
+            'order_id': order.id,
+            'external_reference': order.id,
+            'due_date': order.due_date.strftime('%Y-%m-%d'),  # type: ignore
+            'items': items,
+            'email': order.email,
+            'payable_with': order.payment_method.lower(),
+            'notification_url': postback_url.geturl(),
+            'payer': {
+                'name': order.name,
+                'email': order.email,
+                'cpf_cnpj': order.cnpj if order.cnpj else order.cpf,
+                'address': {
+                    'zip_code': order.address.postcode,
+                    'street': order.address.street,
+                    'number': order.address.street_number,
+                    'district': order.address.neighborhood,
+                    'city': order.address.city,
+                    'state': order.address.state,
+                    'complement': order.address.complement,
+                    'country=': 'Brasil',
+                },
+            },
+        }
+
+        try:
+            response = requests.post(url, json=payload, timeout=15)
+            response.raise_for_status()
+        except requests.HTTPError as err:
+            logger.exception(err)
+            raise
+        else:
+            response = response.json()
+            pix = (
+                Pix(**response['pix'])
+                if order.payment_method == PaymentMethod.PIX
+                else None
+            )
+            bank_slip = (
+                BankSlip(**response['bank_slip'])
+                if order.payment_method == PaymentMethod.BANK_SLIP
+                else None
+            )
+
+            return Invoice(
+                id=response['secure_id'],
+                pdf=bank_slip.bank_slip_url
+                if bank_slip
+                else '%s.pdf' % response['secure_url'],
+                pix=pix,
+            )
+
+    def payment_token(self, credit_card: CreditCard) -> Token:
+        url = self.url(path='/v1/payment_token')
+        payload = {
+            'test': self.credentials.test_mode,
+            'account_id': self.credentials.account_id,
+            'method': 'credit_card',
+            'data': {
+                'number': credit_card.number,
+                'verification_value': credit_card.cvv,
+                'first_name': credit_card.first_name,
+                'last_name': credit_card.last_name,
+                'month': credit_card.exp.strftime('%m'),
+                'year': credit_card.exp.strftime('%Y'),
+            },
+        }
+
+        try:
+            response = requests.post(url, json=payload, timeout=15)
+            response.raise_for_status()
+        except requests.HTTPError as err:
+            logger.exception(err)
+            raise
+        else:
+            return Token(response.json()['id'])
+
+    def charge(
+        self,
+        invoice_id: str,
+        token: Token,
+        installments: int = 1,
+    ) -> Transaction:
+        """
+        O que é Cobrança Direta
+        -----------------------
+        Requisição utilizada para realizar uma cobrança simples de forma pontual
+        utilizando os métodos de pagamento cartão de crédito e boleto.
+
+        Por que usar uma cobrança direta?
+        ---------------------------------
+        Com ela é possível realizar uma cobrança já inserindo os dados do cliente
+        e o token gerado para o cartão de crédito (iugu js).
+        Se o seu intuito é gerar apenas um boleto, essa chamada também retorna
+        o PDF do boleto e o link de pagamento para efetuar o pagamento.
+        """
+
+        url = self.url(path='/v1/charge')
+        payload = {
+            'invoice_id': format_id(invoice_id),
+            'token': token,
+            'months': installments,
+        }
+
+        try:
+            response = requests.post(url, json=payload, timeout=15)
+            response.raise_for_status()
+        except requests.HTTPError as err:
+            logger.exception(err)
+            raise
+        else:
+            success = response.json()['success']
+
+            return Transaction(
+                status=Status.PAID if success else Status.DECLINED,
+                response=response.json(),
+            )
+
+    def get_invoice(self, invoice_id: str) -> dict:
+        url = self.url(path=f'/v1/invoices/{format_id(invoice_id)}')
+
+        try:
+            response = requests.get(url, timeout=15)
+            response.raise_for_status()
+        except requests.HTTPError as err:
+            logger.exception(err)
+            raise
+        else:
+            return response.json()
+
+
+def format_id(invoice_id: str) -> str:
+    return invoice_id.upper().replace('-', '')[:-4]