"""Interface for the ows-users microservice.""" from typing import Optional from uuid import UUID from owsrequest.flask_request import get_ows_headers from owsresponse import response, status as ows_status from notifications.api import ows_client from notifications.constants import error from notifications.constants.service_name import OWS_USERS from notifications.utils.request_params import filter_params def get_user(user_id: str) -> response.Response: """Get a user from ows-users. Args: user_id (str): the orchard user id Returns: response.Response: containing the user. """ response_ = ows_client.get(OWS_USERS, f'/users/{user_id}/minimum-details') if response_.status_code != ows_status.OK: return response.create_error_response( status=response_.status_code, code=error.ERROR_CODE_OWS_USERS_REQUEST, message=response_.text, ) response_data = response_.json() if not response_data: return response.create_not_found_response('No user found') return response.Response(response_data) def get_identity(identity_id: UUID | str) -> response.Response: """Get identity from ows-users. Args: identity_id (UUID | str): the orchard identity id Returns: response.Response: containing the identity. """ response_ = ows_client.get( OWS_USERS, f'/users/identity/{identity_id}', ) if response_.status_code != ows_status.OK: return response.create_error_response( status=response_.status_code, code=error.ERROR_CODE_OWS_USERS_REQUEST, message=response_.text, ) response_data = response_.json() if not response_data: return response.create_not_found_response('No identity found') return response.Response(response_data) def get_applications( identity_id: UUID | str, resource_type: Optional[str] = None, resource_uuid: Optional[str] = None, brand: Optional[str] = None, ) -> response.Response: """Get applications from ows-users. Args: identity_id (UUID | str): the orchard identity id resource_type (Optional[str], optional): the resource type, defaults to None resource_uuid (Optional[str], optional): the resource uuid, defaults to None brand (Optional[str], optional): the brand, defaults to None """ response_ = ows_client.get( OWS_USERS, f'/users/identity/{identity_id}/applications', headers=get_ows_headers(), params=filter_params( resource_type=resource_type, resource_uuid=resource_uuid, brand=brand, ), ) if response_.status_code != ows_status.OK: return response.create_error_response( status=response_.status_code, code=error.ERROR_CODE_OWS_USERS_REQUEST, message=response_.text, ) response_data = response_.json() if not response_data: return response.create_not_found_response('No applications found') return response.Response(response_data['items']) def get_organization_info(org_name: str) -> response.Response: """Get information about an organization. Args: org_name (str): the name of the organization """ response_ = ows_client.get( OWS_USERS, f'/auth0/organizations/{org_name}', headers=get_ows_headers(), ) if response_.status_code != ows_status.OK: return response.create_error_response( status=response_.status_code, code=error.ERROR_CODE_OWS_USERS_REQUEST, message=response_.text, ) response_data = response_.json() if not response_data: return response.create_not_found_response('Organization not found') return response.Response(response_data)