import pprint from dataclasses import asdict, dataclass from uuid import uuid4 from aws_lambda_powertools import Logger from aws_lambda_powertools.utilities.data_classes import ( EventBridgeEvent, event_source, ) from aws_lambda_powertools.utilities.typing import LambdaContext from layercake.dynamodb import DynamoDBPersistenceLayer, KeyChain, KeyPair from boto3clients import dynamodb_client from config import ENROLLMENT_TABLE logger = Logger(__name__) dyn = DynamoDBPersistenceLayer(ENROLLMENT_TABLE, dynamodb_client) @event_source(data_class=EventBridgeEvent) @logger.inject_lambda_context def lambda_handler(event: EventBridgeEvent, context: LambdaContext) -> bool: new_image = event.detail['new_image'] order_id = new_image['id'] enrollments = dyn.collection.query( key=KeyPair(order_id, 'ENROLLMENT#'), ).get('items', []) if not enrollments: items = dyn.collection.get_item( KeyPair(order_id, 'ITEMS'), raise_on_error=False, default=[], ) pprint.pp(items) # docx = { # 'id': f'SEAT#ORG#{org_id}', # 'sk': f'ORDER#{order_id}#ENROLLMENT#{uuid4()}', # 'course': {}, # 'created_at': now_, # } pprint.pp(enrollments) return True # Se houver matriculas # -> com: scheduled_for # -> tenta agendar, se não joga para vagas # -> tenta matriculas, se falhar, joga para vagas # se não houver vagas, gera as vagas. @dataclass(frozen=True) class Course: id: str name: str access_period: int def _get_courses(ids: set) -> tuple[Course, ...]: pairs = tuple(KeyPair(idx, '0') for idx in ids) r = dyn.collection.get_items( KeyChain(pairs), flatten_top=False, ) courses = tuple( Course( id=idx, name=obj['name'], access_period=obj['access_period'], ) for idx, obj in r.items() ) return courses