import json from datetime import datetime from http import HTTPStatus from io import BytesIO from typing import Annotated, Any import requests from aws_lambda_powertools import Logger from aws_lambda_powertools.event_handler.api_gateway import Response, Router from aws_lambda_powertools.event_handler.exceptions import ( BadRequestError, NotFoundError, ) from aws_lambda_powertools.event_handler.openapi.params import Body from layercake.dateutils import now from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair from pydantic import UUID4, BaseModel from api_gateway import JSONResponse from boto3clients import dynamodb_client, s3_client from config import BUCKET_NAME, COURSE_TABLE, PAPERFORGE_API from form_data import parse logger = Logger(__name__) router = Router() dyn = DynamoDBPersistenceLayer(COURSE_TABLE, dynamodb_client) @router.get('/') def get_course(course_id: str): return dyn.collection.get_item( KeyPair(course_id, '0'), exc_cls=NotFoundError, ) class Cert(BaseModel): exp_interval: int | None = None s3_uri: str | None = None def model_dump(self, **kwargs) -> dict[str, Any]: return super().model_dump(exclude_none=True, **kwargs) class Course(BaseModel): id: UUID4 name: str access_period: int cert: Cert rawfile: bytes | None = None @router.put('/') def edit_course(course_id: str): event = router.current_event if not event.decoded_body: raise BadRequestError('Invalid request body') body = BytesIO(event.decoded_body.encode()) course = Course.model_validate( {'id': course_id, 'cert': {}} | parse(event.headers, body), ) now_ = now() if course.rawfile: object_key = f'certs/templates/{course_id}.html' course.cert.s3_uri = f's3://{BUCKET_NAME}/{object_key}' s3_client.put_object( Bucket=BUCKET_NAME, Key=object_key, Body=course.rawfile, ContentType='text/html', ) with dyn.transact_writer() as transact: transact.update( key=KeyPair(str(course.id), '0'), update_expr='SET #name = :name, access_period = :access_period, \ cert = :cert, updated_at = :updated_at', expr_attr_names={ '#name': 'name', }, expr_attr_values={ ':name': course.name, ':cert': course.cert.model_dump(), ':access_period': course.access_period, ':updated_at': now_, }, cond_expr='attribute_exists(sk)', exc_cls=BadRequestError, ) return JSONResponse(HTTPStatus.NO_CONTENT) @router.post('//sample') def sample(course_id: str, s3_uri: Annotated[str, Body(embed=True)]): now_ = now() # Send template URI and data to Paperforge API to generate a PDF r = requests.post( PAPERFORGE_API, data=json.dumps( { 'template_uri': s3_uri, 'args': { 'name': 'Juscelino Kubitschek', 'cpf': '***.810.132-**', 'score': 100, 'started_at': now_.strftime('%d/%m/%Y'), 'completed_at': now_.strftime('%d/%m/%Y'), 'today': _datefmt(now_), 'year': now_.strftime('%Y'), 'expires_at': now_.strftime('%d/%m/%Y'), }, }, ), ) r.raise_for_status() return Response( body=r.content, content_type='application/pdf', status_code=HTTPStatus.OK, headers={ 'Content-Disposition': f'attachment; filename="{course_id}.pdf"', }, ) def _datefmt(dt: datetime) -> str: months = [ 'Janeiro', 'Fevereiro', 'Março', 'Abril', 'Maio', 'Junho', 'Julho', 'Agosto', 'Setembro', 'Outubro', 'Novembro', 'Dezembro', ] return f'{dt.day:02d} de {months[dt.month - 1]} de {dt.year}'