sergio/saladeaula.digital
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip checkout

Browse Source
This commit is contained in:
Sérgio Rafael Siqueira
2026-01-09 11:20:56 -03:00
parent e29e81b253
commit 823134f450
18 changed files with 290 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
orders-events/app/config.py
View File

@@ -5,6 +5,9 @@ ORDER_TABLE: str = os.getenv('ORDER_TABLE') # type: ignore
COURSE_TABLE: str = os.getenv('COURSE_TABLE') # type: ignore
ENROLLMENT_TABLE: str = os.getenv('ENROLLMENT_TABLE') # type: ignore
IUGU_ACCOUNT_ID: str = 'AF01CF1B3451459F92666F10589278EE'
IUGU_API_TOKEN: str = os.getenv('IUGU_API_TOKEN') # type: ignore
BUCKET_NAME: str = os.getenv('BUCKET_NAME') # type: ignore
EMAIL_SENDER = ('EDUSEG®', 'noreply@eduseg.com.br')

17
orders-events/app/events/append_org_id.py
View File

@@ -57,14 +57,15 @@ def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool:
},
)
transact.update(
key=KeyPair(new_image['id'], 'author'),
update_expr='SET user_id = :user_id, updated_at = :updated_at',
expr_attr_values={
':user_id': r['user_id'],
':updated_at': now_,
},
)
if 'org_id' not in new_image:
transact.update(
key=KeyPair(new_image['id'], 'author'),
update_expr='SET user_id = :user_id, updated_at = :updated_at',
expr_attr_values={
':user_id': r['user_id'],
':updated_at': now_,
},
)
logger.info('IDs updated')

135
orders-events/app/iugu.py
View File

@@ -16,36 +16,66 @@ Documentation:
"""
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
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
from layercake.extra_types import CreditCard
from pydantic import BaseModel, ConfigDict, HttpUrl
logger = Logger(__name__)
class Status(Enum):
class Status(str, Enum):
PAID = 'PAID'
DECLINED = 'DECLINED'
class PaymentMethod(Enum):
class PaymentMethod(str, Enum):
PIX = 'PIX'
BANK_SLIP = 'BANK_SLIP'
CREDIT_CARD = 'CREDIT_CARD'
class Order: ...
@dataclass
class Address(BaseModel):
postcode: str
neighborhood: str
city: str
state: str
address1: str
address2: str | None = None
class CreditCard: ...
class Item(BaseModel):
id: str
name: str
quantity: int = 1
unit_price: Decimal
class Order(BaseModel):
model_config = ConfigDict(use_enum_values=True)
id: str
email: str
name: str
due_date: datetime
address: Address
items: tuple[Item, ...]
payment_method: PaymentMethod
cpf: str | None = None
cnpj: str | None = None
@dataclass
class Invoice: ...
@dataclass
class BankSlip(BaseModel):
digitable_line: str
bank_slip_url: HttpUrl
@@ -115,7 +145,7 @@ class Iugu:
payload = {
'order_id': order.id,
'external_reference': order.id,
'due_date': order.due_date.strftime('%Y-%m-%d'), # type: ignore
'due_date': order.due_date.strftime('%Y-%m-%d'),
'items': items,
'email': order.email,
'payable_with': order.payment_method.lower(),
@@ -126,68 +156,71 @@ class Iugu:
'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,
'street': order.address.address1,
'number': '',
'district': order.address.neighborhood,
'city': order.address.city,
'state': order.address.state,
'complement': order.address.complement,
'complement': order.address.address2,
'country=': 'Brasil',
},
},
}
try:
response = requests.post(url, json=payload, timeout=15)
response.raise_for_status()
r = requests.post(url, json=payload, timeout=15)
r.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 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,
)
# 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()
r = requests.post(
url,
json={
'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_month,
'year': credit_card.exp_year,
},
},
timeout=15,
)
r.raise_for_status()
except requests.HTTPError as err:
logger.exception(err)
raise
else:
return Token(response.json()['id'])
return Token(r.json()['id'])
def charge(
self,
@@ -234,13 +267,13 @@ class Iugu:
url = self.url(path=f'/v1/invoices/{format_id(invoice_id)}')
try:
response = requests.get(url, timeout=15)
response.raise_for_status()
r = requests.get(url, timeout=15)
r.raise_for_status()
except requests.HTTPError as err:
logger.exception(err)
raise
else:
return response.json()
return r.json()
def format_id(invoice_id: str) -> str: