This commit is contained in:
2026-01-12 22:36:34 -03:00
parent 6fb00c24c1
commit 29b8305713
15 changed files with 327 additions and 138 deletions

View File

@@ -15,6 +15,7 @@ Documentation:
- https://support.iugu.com/hc/pt-br/articles/212456346-Usar-cart%C3%B5es-de-teste-em-modo-de-teste
"""
import os
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
@@ -24,7 +25,10 @@ from urllib.parse import ParseResult, urlparse
import requests
from aws_lambda_powertools import Logger
from layercake.extra_types import CreditCard
from pydantic import BaseModel, ConfigDict, HttpUrl
from pydantic import BaseModel, ConfigDict
HTTP_CONNECT_TIMEOUT = int(os.environ.get('HTTP_CONNECT_TIMEOUT', 1))
HTTP_READ_TIMEOUT = int(os.environ.get('HTTP_READ_TIMEOUT', 3))
logger = Logger(__name__)
@@ -71,16 +75,6 @@ class Order(BaseModel):
cnpj: str | None = None
@dataclass
class Invoice: ...
@dataclass
class BankSlip(BaseModel):
digitable_line: str
bank_slip_url: HttpUrl
@dataclass
class Credentials:
account_id: str
@@ -88,17 +82,6 @@ class Credentials:
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')
@@ -114,7 +97,7 @@ class Iugu:
self,
order: Order,
postback_url: ParseResult | str,
) -> Invoice:
) -> dict:
"""
O que é uma fatura?
-------------------
@@ -127,6 +110,8 @@ class Iugu:
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.
- https://dev.iugu.com/reference/criar-fatura
"""
url = self.url(path='/v1/invoices')
@@ -168,33 +153,25 @@ class Iugu:
}
try:
r = requests.post(url, json=payload, timeout=15)
r = requests.post(
url, json=payload, timeout=(HTTP_CONNECT_TIMEOUT, HTTP_READ_TIMEOUT)
)
r.raise_for_status()
except requests.HTTPError as err:
logger.exception(err)
raise
else:
return r.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) -> dict:
"""When creating a invoice, it can't make the charge immediately the invoice.
It's necessary make have a token (payment token) to charge.
Payment token doesn't depends an invoice, just a credit card to charge later.
- https://dev.iugu.com/reference/criar-token
"""
def payment_token(self, credit_card: CreditCard) -> Token:
url = self.url(path='/v1/payment_token')
try:
@@ -213,21 +190,21 @@ class Iugu:
'year': credit_card.exp_year,
},
},
timeout=15,
timeout=(HTTP_CONNECT_TIMEOUT, HTTP_READ_TIMEOUT),
)
r.raise_for_status()
except requests.HTTPError as err:
logger.exception(err)
raise
else:
return Token(r.json()['id'])
return r.json()
def charge(
self,
invoice_id: str,
token: Token,
token: str,
installments: int = 1,
) -> Transaction:
) -> dict:
"""
O que é Cobrança Direta
-----------------------
@@ -240,6 +217,8 @@ class Iugu:
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.
- https://dev.iugu.com/reference/cobranca-direta
"""
url = self.url(path='/v1/charge')
@@ -250,24 +229,21 @@ class Iugu:
}
try:
response = requests.post(url, json=payload, timeout=15)
response.raise_for_status()
r = requests.post(
url, json=payload, timeout=(HTTP_CONNECT_TIMEOUT, HTTP_READ_TIMEOUT)
)
r.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(),
)
return r.json()
def get_invoice(self, invoice_id: str) -> dict:
url = self.url(path=f'/v1/invoices/{format_id(invoice_id)}')
try:
r = requests.get(url, timeout=15)
r = requests.get(url, timeout=(HTTP_CONNECT_TIMEOUT, HTTP_READ_TIMEOUT))
r.raise_for_status()
except requests.HTTPError as err:
logger.exception(err)