89 lines
2.1 KiB
Python
89 lines
2.1 KiB
Python
from decimal import Decimal
|
|
from typing import Annotated
|
|
|
|
from aws_lambda_powertools import Logger
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from aws_lambda_powertools.event_handler.exceptions import (
|
|
NotFoundError,
|
|
)
|
|
from aws_lambda_powertools.event_handler.openapi.params import Body
|
|
from layercake.batch import BatchProcessor
|
|
from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair
|
|
from layercake.extra_types import CnpjStr, CpfStr, NameStr
|
|
from pydantic import UUID4, BaseModel, EmailStr, FutureDate
|
|
|
|
from boto3clients import dynamodb_client
|
|
from config import ENROLLMENT_TABLE
|
|
from middlewares.authentication_middleware import User as Authenticated
|
|
|
|
logger = Logger(__name__)
|
|
router = Router()
|
|
dyn = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client)
|
|
processor = BatchProcessor()
|
|
|
|
|
|
class SubscriptionNotFoundError(NotFoundError): ...
|
|
|
|
|
|
class User(BaseModel):
|
|
id: str | UUID4
|
|
name: NameStr
|
|
cpf: CpfStr
|
|
email: EmailStr
|
|
|
|
|
|
class Course(BaseModel):
|
|
id: UUID4
|
|
name: str
|
|
access_period: int
|
|
unit_price: Decimal
|
|
|
|
|
|
class Enrollment(BaseModel):
|
|
user: User
|
|
course: Course
|
|
scheduled_for: FutureDate | None = None
|
|
|
|
|
|
class Org(BaseModel):
|
|
id: str | UUID4
|
|
name: str
|
|
cnpj: CnpjStr
|
|
|
|
|
|
@router.post('/')
|
|
def enroll(
|
|
org_id: Annotated[UUID4 | str, Body(embed=True)],
|
|
enrollments: Annotated[tuple[Enrollment, ...], Body(embed=True)],
|
|
):
|
|
created_by: Authenticated = router.context['user']
|
|
org = dyn.collection.get_items(
|
|
KeyPair(
|
|
pk=str(org_id),
|
|
sk='0',
|
|
)
|
|
+ KeyPair(
|
|
pk='SUBSCRIPTION',
|
|
sk=f'ORG#{org_id}',
|
|
rename_key='subscription',
|
|
)
|
|
)
|
|
|
|
subscribed = 'subscription' in org
|
|
if not subscribed:
|
|
return checkout(Org.model_validate(org), enrollments, created_by=created_by)
|
|
|
|
scheduled, unscheduled = [], []
|
|
for x in enrollments:
|
|
(scheduled if x.scheduled_for else unscheduled).append(x)
|
|
|
|
print(scheduled, created_by)
|
|
|
|
|
|
def checkout(
|
|
org: Org,
|
|
enrollments: tuple[Enrollment, ...],
|
|
created_by: Authenticated,
|
|
):
|
|
print(org, enrollments, created_by)
|