"""Interface to the ows-users microservice.""" from oto import response from owsrequest import request from collaborator.constants import error, service_name from collaborator.utils import owsrequest from collaborator.utils.typing import Account, User def get_master_contact(account: Account) -> response.Response: """Get the master contact for a label. Args: account (Account): Account Tuple. Returns: oto.response.Response """ master_contact_url = f"/ows/{account.type}/{account.id}/primary-contact" params = {"active": "Y"} res = request.get(service_name.OWS_USERS, master_contact_url, params=params) return owsrequest.unwrap_data_or_error(res, error.ERROR_CODE_OWS_USERS) def get_user_metadata(user: User) -> response.Response: """Get user metadata. Args: user (User): The User tuple Returns: oto.response.Response """ user_id = str(user) user_metadata_url = f"/users/{user_id}" res = request.get(service_name.OWS_USERS, user_metadata_url) return owsrequest.unwrap_data_or_error(res, error.ERROR_CODE_OWS_USERS) def get_identity_metadata(user: User) -> response.Response: """Get identity metadata. Args: user (User): The User tuple Returns: oto.response.Response """ identity_id = user.id identity_metadata_url = f"/users/identity/{identity_id}" res = request.get(service_name.OWS_USERS, identity_metadata_url) return owsrequest.unwrap_data_or_error(res, error.ERROR_CODE_OWS_USERS) def get_profile(profile_id: int, profile_type: str) -> response.Response: """Get profile. Args: profile_id (int): The Profile ID. profile_type (str): The Profile type. Returns: oto.response.Response """ profile_url = f"/profile/profile_id/{profile_id}/profile_type/{profile_type}" res = request.get(service_name.OWS_USERS, profile_url) return owsrequest.unwrap_data_or_error(res, error.ERROR_CODE_OWS_USERS) def get_profile_for_identity(profile_type: str, identity_id: str): """Get a Profile profile_type for user with identity_id. Args: profile_type (str): The Profile type. identity_id (str): the id of the user. Returns: response.Response: the data of the model. """ resource_url = f"/users/identity/{identity_id}/profiles" res = request.get(service_name.OWS_USERS, resource_url) if res.status_code != 200: return None profiles = res.json().get("items", []) profiles = [p for p in profiles if p.get("profile_type") == profile_type] profile = next(iter(profiles), None) return profile