155 lines
3.8 KiB
Python
155 lines
3.8 KiB
Python
import re
|
|
from decimal import Decimal
|
|
from http import HTTPStatus
|
|
from typing import Any, Literal
|
|
from uuid import uuid4
|
|
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from layercake.dateutils import now
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer
|
|
from layercake.extra_types import CnpjStr, CpfStr, NameStr
|
|
from pydantic import (
|
|
UUID4,
|
|
BaseModel,
|
|
ConfigDict,
|
|
EmailStr,
|
|
Field,
|
|
field_validator,
|
|
model_validator,
|
|
)
|
|
|
|
from api_gateway import JSONResponse
|
|
from boto3clients import dynamodb_client
|
|
from config import ORDER_TABLE
|
|
from routes.enrollments.enroll import Enrollment
|
|
from routes.orgs.address import address
|
|
|
|
router = Router()
|
|
dyn = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client)
|
|
|
|
|
|
class User(BaseModel):
|
|
id: UUID4 | str
|
|
name: NameStr
|
|
|
|
|
|
class Address(BaseModel):
|
|
model_config = ConfigDict(str_strip_whitespace=True)
|
|
|
|
postcode: str
|
|
address1: str
|
|
address2: str | None = None
|
|
neighborhood: str
|
|
city: str
|
|
state: str
|
|
|
|
@field_validator('postcode')
|
|
@classmethod
|
|
def ensure_numbers(cls, v: str) -> str:
|
|
return re.sub(r'\D', '', v)
|
|
|
|
|
|
class Item(BaseModel):
|
|
id: UUID4
|
|
name: str
|
|
unit_price: Decimal
|
|
quantity: int = 1
|
|
|
|
|
|
class Coupon(BaseModel):
|
|
code: str
|
|
type: Literal['PERCENT', 'FIXED']
|
|
amount: Decimal
|
|
|
|
|
|
class Checkout(BaseModel):
|
|
model_config = ConfigDict(str_strip_whitespace=True)
|
|
|
|
id: UUID4 = Field(default_factory=uuid4)
|
|
name: str
|
|
email: EmailStr
|
|
address: Address
|
|
payment_method: Literal['PIX', 'CREDIT_CARD', 'BANK_SLIP', 'MANUAL']
|
|
items: tuple[Item, ...]
|
|
enrollments: tuple[Enrollment, ...] | None = None
|
|
coupon: Coupon | None = None
|
|
org_id: UUID4 | str | None = None
|
|
user_id: UUID4 | str | None = None
|
|
cnpj: CnpjStr | None = None
|
|
cpf: CpfStr | None = None
|
|
created_by: User | None = None
|
|
|
|
@model_validator(mode='after')
|
|
def verify_fields(self):
|
|
if not any([self.cnpj, self.cpf]):
|
|
raise ValueError('cnpj or cpf is required')
|
|
|
|
if self.cnpj is not None:
|
|
if self.org_id is None:
|
|
raise ValueError('org_id is missing')
|
|
|
|
if self.created_by is None:
|
|
raise ValueError('created_by is missing')
|
|
|
|
if self.cpf is not None and self.user_id is None:
|
|
raise ValueError('user_id is missing')
|
|
|
|
return self
|
|
|
|
def model_dump(self, **kwargs) -> dict[str, Any]:
|
|
return super().model_dump(
|
|
exclude_none=True,
|
|
exclude={'items', 'address', 'created_by'},
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
@router.post('/')
|
|
def checkout(payload: Checkout):
|
|
now_ = now()
|
|
order_id = str(payload.id)
|
|
address = payload.address
|
|
coupon = payload.coupon
|
|
|
|
with dyn.transact_writer() as transact:
|
|
transact.put(
|
|
item={
|
|
'id': order_id,
|
|
'sk': '0',
|
|
'total': '',
|
|
'discount': '',
|
|
'due_date': '',
|
|
'created_at': now_,
|
|
}
|
|
| ({'coupon': coupon.code} if coupon else {})
|
|
| payload.model_dump()
|
|
)
|
|
transact.put(
|
|
item={
|
|
'id': order_id,
|
|
'sk': 'ITEMS',
|
|
'items': [],
|
|
'created_at': now_,
|
|
}
|
|
)
|
|
transact.put(
|
|
item={
|
|
'id': order_id,
|
|
'sk': 'ADDRESS',
|
|
'created_at': now_,
|
|
}
|
|
| address.model_dump()
|
|
)
|
|
|
|
if coupon:
|
|
transact.put(
|
|
item={
|
|
'id': order_id,
|
|
'sk': 'COUPON',
|
|
'created_at': now_,
|
|
}
|
|
| coupon.model_dump()
|
|
)
|
|
|
|
return JSONResponse(body={'id': order_id}, status_code=HTTPStatus.CREATED)
|