from datetime import date from typing import Annotated from aws_lambda_powertools.event_handler.api_gateway import Router from aws_lambda_powertools.event_handler.openapi.params import Path, Query from layercake.dynamodb import DynamoDBPersistenceLayer, KeyPair from boto3clients import dynamodb_client from config import ORDER_TABLE, USER_TABLE router = Router() order_layer = DynamoDBPersistenceLayer(ORDER_TABLE, dynamodb_client) user_layer = DynamoDBPersistenceLayer(USER_TABLE, dynamodb_client) @router.get('/', compress=True, tags=['Billing']) def get_billing( id: Annotated[str, Path()], start_date: Annotated[date, Query()], end_date: Annotated[date, Query()], ): result = order_layer.collection.query( KeyPair( pk=f'BILLING#ORG#{id}', sk=f'START#{start_date}#END#{end_date}', ), limit=100, ) return result @router.get('//terms', compress=True, tags=['Billing']) def get_terms(id: Annotated[str, Path()]): return user_layer.collection.get_item( KeyPair( pk=id, sk='metadata#billing_policy', # Post-migration: uncomment the following line # sk='METADATA#BILLING_TERMS', ), raise_on_error=False, default={}, )