from http import HTTPStatus from aws_lambda_powertools.event_handler import Response, content_types from aws_lambda_powertools.event_handler.api_gateway import Router from aws_lambda_powertools.event_handler.exceptions import ( BadRequestError, ) from layercake.dynamodb import ( DynamoDBCollection, DynamoDBPersistenceLayer, SortKey, TransactKey, ) from pydantic.main import BaseModel from typing_extensions import Literal from boto3clients import dynamodb_client from org import update_policies from settings import USER_TABLE router = Router() org_layer = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client) org_collect = DynamoDBCollection(org_layer, exception_cls=BadRequestError) @router.get( '//policies', compress=True, tags=['Organization'], summary='Get organization policies', ) def get_policies(id: str): return org_collect.get_items( TransactKey(id) + SortKey('billing_policy') + SortKey('payment_policy'), flatten_top=False, ) class BillingPolicy(BaseModel): billing_day: int payment_method: Literal['PIX', 'BANK_SLIP', 'MANUAL'] class PaymentPolicy(BaseModel): due_days: int class Policies(BaseModel): billing_policy: BillingPolicy | None = None payment_policy: PaymentPolicy | None = None @router.put('//policies', compress=True, tags=['Organization']) def put_policies(id: str, payload: Policies): payment_policy = payload.payment_policy billing_policy = payload.billing_policy update_policies( id, payment_policy=payment_policy.model_dump() if payment_policy else {}, billing_policy=billing_policy.model_dump() if billing_policy else {}, persistence_layer=org_layer, ) return Response( body=payload, content_type=content_types.APPLICATION_JSON, status_code=HTTPStatus.OK, )