"""User based helpers methods.""" from flask import g, request from abacus_common_logic.constants.constants import ( APPLICATION_PROFILE_TYPE, DEFAULT_USER_ID, DEFAULT_USER_TYPE, IDENTITY_HEADER, USER_HEADER_DELIMITER, USER_HEADER_NAME, ) def set_flask_user_details_from_headers(): """Parse user information in the headers and add it to flask.g. TODO: Add logic here to restrict anonymous access """ orchard_identity_id = request.headers.get(IDENTITY_HEADER, None) grass_user_details = request.headers.get(USER_HEADER_NAME, None) user_type, user_id = parse_user_headers(orchard_identity_id, grass_user_details) g.user_details = {'type': user_type, 'id': user_id} def parse_user_headers(orchard_identity_id, grass_user_details=None): """Get the user_id and type from the request headers.""" user_type = DEFAULT_USER_TYPE user_id = DEFAULT_USER_ID if orchard_identity_id: user_type = APPLICATION_PROFILE_TYPE user_id = orchard_identity_id elif grass_user_details: user_type, user_id = get_user_details_from_string(grass_user_details) return user_type, user_id def get_user_details_from_string(user_details): """Parse the pattern 'user_type:user_id' into a tuple.""" user_type = DEFAULT_USER_TYPE user_id = DEFAULT_USER_ID if user_details: header_pieces = user_details.split(USER_HEADER_DELIMITER) if len(header_pieces) == 2: user_id = header_pieces.pop() or user_id user_type = header_pieces.pop() or user_type return user_type, user_id def get_flask_user_id(): """Return application level user id.""" if not g: return DEFAULT_USER_ID if ( hasattr(g, 'user_details') and isinstance(g.user_details, dict) and g.user_details.get('id') ): return g.user_details['id'] if hasattr(g, 'request_context') and getattr( g.request_context, 'jwt_identity_id', False ): return g.request_context.jwt_identity_id return DEFAULT_USER_ID