"""API utilities.""" import functools from typing import Any, Callable, Literal, TypeVar, cast import sentry_sdk import stringcase from flask import Response as FlaskResponse, g, request from marshmallow import Schema, ValidationError from owsrequest.context import RequestContext from owsresponse import response from owsresponse.adaptors.flask import flaskify from notifications.constants import error from notifications.constants.header import ORCHARD_SYST_IDENTITY_ID DataSource = Literal['json'] | Literal['args'] Handler = TypeVar('Handler', bound=Callable[..., FlaskResponse]) def validate_request_data( schema: Schema, partial: bool = False, data_source: DataSource = 'json' ) -> Callable[[Handler], Handler]: """Decorate requests' input data validation. Args: schema (marshmallow.Schema()): schema instance to apply partial (bool): validate only derived fields data_source (DataSource): where to get data from """ def validator(func: Handler) -> Handler: @functools.wraps(func) def wrapper(*args: Any, **kwargs: Any) -> FlaskResponse: try: validated = schema.load(getattr(request, data_source), partial=partial) except ValidationError as err: sentry_sdk.capture_exception() return flaskify( response.create_error_response( code=error.ERROR_CODE_VALIDATION_ERROR, message=err.messages ) ) else: kwargs.update({f'validated_{data_source}': validated}) return func(*args, **kwargs) return cast(Handler, wrapper) return validator def to_snake(data: dict[str, Any]) -> dict[str, Any]: """Transform the keys of the payload to snake case. Args: data (dict): the payload to format the keys. Returns: dict: a new dictionary with the formatted payload """ snake_case_dict = {} for k, v in data.items(): snake_case_dict[stringcase.snakecase(k)] = v return snake_case_dict def to_camel(data: dict[str, Any]) -> dict[str, Any]: """Transform the keys of the payload to camel case. Args: data (dict): the payload to format the keys. Returns: dict: a new dictionary with the formatted payload """ camel_case_dict = {} for k, v in data.items(): camel_case_dict[stringcase.camelcase(k)] = v return camel_case_dict def is_orchard_syst(context: RequestContext) -> bool: """Determine whether the request parameters represent Orchard System user.""" return context.identity_id == ORCHARD_SYST_IDENTITY_ID def jwt_check(func: Callable[..., Any]) -> Callable[..., Any]: """Check for a valid jwt.""" @functools.wraps(func) def decorator(**kwargs: Any) -> FlaskResponse: identity_uuid = g.request_context.jwt_identity_id if not identity_uuid: return flaskify( response.create_error_response( error.ERROR_CODE_AUTHORIZATION, message='Request context has no identity uuid.', status=401, ) ) return func(**kwargs) return decorator