"""Workstation User. The Grass implementation of a user is a little different from the one in VAPI. Users in Grass are resource owners, in this perspective, the user id represent the resource they own. VAPI uses inconsistenly the ``user_id``, ``vendor.id`` and ``subaccount.id``. In this approach, we virtually flatten the object to work consistently. You can still – if needed – access the ``vendor_id``, ``contact_id``, and ``subaccount_id``. The structure: id (int): the user id. Corresponds to the ``contact_vendor.id`` field. This id should be used instead of the default `id` whenever possible. entity_id (int): Corresponds to either the vendor id or the subaccount id (depending on the account type.) type (str): the account type (S if it's a subaccount, L if it's a label.) Warning: More data is available in the table but not directly accessible here. As we are moving more and more data over to grass, we will need to update this table accordingly to what is in the table (and used from the table.) """ from ddtrace import tracer from sqlalchemy import Column, Integer, String from grass import config from grass.connectors import microservices, mysql from grass.consts import resource, user from grass.utils import response class WorkstationUser(mysql.BaseModel): """Workstation User.""" __tablename__ = 'vend_contact' id = Column(Integer, primary_key=True) # noqa vendor_id = Column(Integer) contact_id = Column(Integer) subaccount_id = Column(Integer) orchard_identity_id = Column('auth0_user_id', String) roles = [] orchard_identity_uuid = None @property def user_id(self): """Return the user id. The user id is a unique identifier across our different systems. It is composed of the client name (here ALW), and the unique id for this user. Return: str: the user id. """ return f'{user.UserGroups.ALW}:{self.id}' @property def user_group(self): """Return the user group. Return: str: the user id. """ return user.UserGroups.ALW @property def account_id(self): """Return the account id. Warning: The `entity_id` is different from the `id`. The `id` corresponds to the unique identifier of the user, whereas the `entity_id` corresponds to the resource owner id. Returns: int: the entity id. """ if self.subaccount_id: return self.subaccount_id else: return self.vendor_id @property def account_type(self): """Standard account type. The table is inconsistently populated: some accounts are for labels, some other accounts are for label sub accounts. This method provides the type of account. Please note: Labels / Distributors are vendors. """ if self.subaccount_id: return user.AccountTypes.SUBACCOUNT else: return user.AccountTypes.VENDOR def set_roles(self, roles): """Sets roles attribute for a ALW user. Converts role names to lowercase. Args: roles (list): list of ALW roles in text format ex. ['catalog', 'accounting'] """ self.roles = [] for role in roles: self.roles.append(str.lower(role)) def set_identity_uuid(self, identity_uuid): self.orchard_identity_uuid = identity_uuid def set_identity_id(self, identity_id): self.orchard_identity_id = identity_id @tracer.wrap() def get_workstation_user_by_id(user_id): """Get a workstation user by its id. Args: user_id (str): the user's id. Returns: Response: the user's information. """ user_id = int(user_id.replace('alw:', '')) session = mysql.session() row = session.query(WorkstationUser).filter(WorkstationUser.id == user_id).first() session.close() if row: return response.Response(row) return response.create_not_found_response() @tracer.wrap() def get_linked_account_details(user_id, auth0_user_id): """Fetch all accounts that are linked to this user. Args: user_id (int): vend_contact.id. auth0_user_id (str): the auth0 user id. Returns: dict: A list of dict with account details. """ session = mysql.session() sql = f"""SELECT vc.id as vc_id, v.vendor_id, vc.subaccount_id, s.subaccount_name, v.label_identifier, v.name as vendor_name, v.company, vc.auth0_user_id, vc.auth0_primary FROM vend_contact vc INNER JOIN vendor v ON v.vendor_id = vc.vendor_id LEFT JOIN subaccount s ON vc.subaccount_id = s.subaccount_id WHERE vc.id != {user_id} AND vc.active = "Y" AND vc.auth0_user_id = '{auth0_user_id}'""" vend_response = session.execute(sql) result = vend_response.fetchall() if result: result_obj = [resource.LinkedAccountResource(*data) for data in result] return response.Response(message=result_obj) return response.Response(message=[]) @tracer.wrap() def create_workstation_user(**data): """Create workstation user based on a set of data. Args: data (dict): the data related to the user. Returns: Response: the user information and the status of the request. If an error has happened when commiting the information, a 500 is returned. """ user = WorkstationUser(**data) session = mysql.session() try: session.add(user) session.commit() except: session.rollback() return response.create_fatal_response(dict(error='Unable to create the user')) session.close() return response.Response(user) @tracer.wrap() def get_resource_privileges_for_user(user_id, resource_name): """Get resource privileges for user. Args: user_id (int): the user's id. resource_name (str): the name of the resource. Returns: Response: details of the resource. """ user_id = user_id.replace('alw:', '') connection = mysql.engine.raw_connection() results = [] try: cursor = connection.cursor() cursor.callproc(resource.DB_PRIVILEGES_PROCEDURE_NAME, [user_id, resource_name]) results = [resource.Resource(*data) for data in cursor.fetchall()] cursor.close() finally: connection.close() return results @tracer.wrap() def get_roles_for_user(user_id): """Get vend_contact_roles for ALW user. Args: user_id (str): alw:vend_contact_id. Returns: Response: message payload is a list of all role_ids for user. """ user_id = user_id.replace('alw:', '') roles = dict(role_ids=[], role_names=[]) session = mysql.session() try: # get user roles sql = """SELECT role_id, role FROM vend_contact_roles vcr JOIN vendor_roles vr on vcr.role_id = vr.id WHERE vend_contact_id = :user_id""" roles_result = session.execute(sql, {'user_id': user_id}) all_roles = roles_result.fetchall() # look for any vendor level feature control restricts # that affect roles sql = f"""SELECT feature_id FROM vendor_restricted_features vsf JOIN vend_contact vc on vsf.`vendor_id` = vc.vendor_id WHERE id = :user_id AND feature_id in ( {user.MARKETING_FEATURE_CONTROL}, {user.ANALYTICS_FEATURE_CONTROL} )""" restricted_feature_result = session.execute(sql, {'user_id': user_id}) restricted_feature_result = restricted_feature_result.fetchall() restricted_features = [f for (f,) in restricted_feature_result] if all_roles: for role in all_roles: # skip analytics role if feature is restricted for vendor if ( role[0] == 3 and user.ANALYTICS_FEATURE_CONTROL in restricted_features ): continue # skip marketing role if feature is restricted for vendor if ( role[0] == 2 and user.MARKETING_FEATURE_CONTROL in restricted_features ): continue roles['role_ids'].append(role[0]) roles['role_names'].append(role[1]) finally: session.close() return response.Response(message=roles) @tracer.wrap() def get_user_identity(user_id): """Get identity info from ows-users. Args: user_id (int): Workstation user id. Returns: Response: indentity object """ user_id = user_id.replace('alw:', '') result = microservices.request( 'GET', 'ows-users', f'/profile/profile_id/{user_id}/' f'profile_type/LabelProfile/identity', ) if result.status_code == 200: return response.Response(result.json()) else: try: errors = result.json() except ValueError: errors = result.text return response.create_error_response(errors=errors, status=result.status_code) if config.environment == config.TEST_ENVIRONMENT: WorkstationUser.__table__.create(mysql.engine)