"""Utils module providing helper functions around request parameters.""" from typing import Any, Collection from flask import Request as FlaskRequest from owsrequest import flask_request from owsresponse import response, status from notifications.constants import error from notifications.constants.header import ORCHARD_USER_ID from notifications.constants.stream import DEFAULT_USER_FEED, USER_FEEDS def get_data_params(request: FlaskRequest) -> dict[str, Any]: """Parse and return the request body params. Args: request (Flask.Request): the request object Returns: dict: containing the body params """ return request.get_json(silent=True, force=True) or {} def get_header_params(request: FlaskRequest) -> dict[str, str | None]: """Parse and return the request header params. Args: request (Flask.Request): the request object Returns: dict: containing the header params """ account_type, account_id = flask_request.get_grass_headers(request) if account_type and account_id: feed_id = '{0}_{1}'.format(account_type, account_id) else: feed_id = None return {ORCHARD_USER_ID: request.headers.get(ORCHARD_USER_ID, None), 'feed_id': feed_id} def validate_params(params: dict[str, Any], keys: Collection[str]) -> response.Response: """Extract and validate a value in params for each keys. Args: params (dict): the parameters dict keys ([str]): the list of required keys Returns: response.Response: 400 or 200 containing the list of values """ validated_params = [] for key in keys: if key not in params or not params[key]: return response.create_error_response( status=status.BAD_REQUEST, code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_MISSING_PARAMETERS, ) validated_params.append(params[key]) return response.Response(validated_params) def validate_user_feed_name(request: FlaskRequest) -> response.Response: """Extract and validate the user feed name from the query parameters. Args: request (Flask.Request): the request object Returns: response.Response: 400 or 200 containing the user feed name """ user_feed_name = request.args.get('user_feed_name', DEFAULT_USER_FEED) if user_feed_name and user_feed_name not in USER_FEEDS: return response.create_error_response( status=status.BAD_REQUEST, code=error.ERROR_CODE_BAD_REQUEST, message=error.ERROR_MESSAGE_INVALID_USER_FEED_NAME, ) return response.Response(user_feed_name) def filter_params(**params: Any) -> dict[str, Any]: """Remove None values (with keys) from params. Args: **params: Arbitrary keyword arguments to be filtered. Any key with a value of None will be excluded from the result. Returns: dict[str, Any]: A new dictionary containing only the entries from params where the value was not None. """ return {k: v for k, v in params.items() if v is not None}