64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
from typing import Any
|
|
|
|
from aws_lambda_powertools import Logger, Tracer
|
|
from aws_lambda_powertools.event_handler import Response, content_types
|
|
from aws_lambda_powertools.event_handler.api_gateway import (
|
|
APIGatewayHttpResolver,
|
|
)
|
|
from aws_lambda_powertools.event_handler.exceptions import ServiceError
|
|
from aws_lambda_powertools.logging import correlation_paths
|
|
from aws_lambda_powertools.utilities.typing import LambdaContext
|
|
|
|
from routes.authentication import router as authentication
|
|
from routes.authorize import router as authorize
|
|
from routes.forgot import router as forgot
|
|
from routes.jwks import router as jwks
|
|
from routes.lookup import router as lookup
|
|
from routes.openid_configuration import router as openid_configuration
|
|
from routes.register import router as register
|
|
from routes.reset import router as reset
|
|
from routes.revoke import router as revoke
|
|
from routes.token import router as token
|
|
from routes.userinfo import router as userinfo
|
|
|
|
logger = Logger(__name__)
|
|
tracer = Tracer()
|
|
app = APIGatewayHttpResolver(enable_validation=True)
|
|
app.include_router(authentication)
|
|
app.include_router(authorize)
|
|
app.include_router(forgot)
|
|
app.include_router(jwks)
|
|
app.include_router(lookup)
|
|
app.include_router(openid_configuration)
|
|
app.include_router(register)
|
|
app.include_router(reset)
|
|
app.include_router(revoke)
|
|
app.include_router(token)
|
|
app.include_router(userinfo)
|
|
|
|
|
|
@app.get('/health')
|
|
def health():
|
|
return {'status': 'available'}
|
|
|
|
|
|
@app.exception_handler(ServiceError)
|
|
def exc_error(exc: ServiceError):
|
|
logger.exception(exc)
|
|
|
|
return Response(
|
|
body={
|
|
'type': type(exc).__name__,
|
|
'message': str(exc),
|
|
},
|
|
content_type=content_types.APPLICATION_JSON,
|
|
status_code=exc.status_code,
|
|
compress=True,
|
|
)
|
|
|
|
|
|
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_HTTP)
|
|
@tracer.capture_lambda_handler
|
|
def lambda_handler(event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
|
|
return app.resolve(event, context)
|