92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
from http import HTTPStatus
|
|
from io import BytesIO
|
|
from typing import Any
|
|
|
|
from aws_lambda_powertools import Logger
|
|
from aws_lambda_powertools.event_handler.api_gateway import Router
|
|
from aws_lambda_powertools.event_handler.exceptions import (
|
|
BadRequestError,
|
|
NotFoundError,
|
|
)
|
|
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
|
|
from form_data import parse
|
|
|
|
logger = Logger(__name__)
|
|
router = Router()
|
|
dyn = DynamoDBPersistenceLayer(COURSE_TABLE, dynamodb_client)
|
|
|
|
|
|
@router.get('/<course_id>')
|
|
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('/<course_id>')
|
|
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/{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)
|