"""Helper module for context functions.""" from owsrequest import error_response from owsrequest.constants import headers from owsresponse import response from features.constants import error from features.constants import context as context_constants from features.models import ows_users class Context: """Class to organize context logic and validations.""" def __init__( self, orchard_user_id=None, subaccount_id=None, vendor_id=None, identity_id=None, identity_uuid=None, ): """Initialize class.""" self.orchard_user_id = str(orchard_user_id) if orchard_user_id else None self.subaccount_id = str(subaccount_id) if subaccount_id else None self.vendor_id = str(vendor_id) if vendor_id else None self.identity_uuid = str(identity_uuid) if identity_uuid else None self.identity_id = str(identity_id) if identity_id else None def validate(self): """Validate initialization arguments.""" # check that accounts have users if (self.subaccount_id and not self.orchard_user_id) or ( self.vendor_id and not self.orchard_user_id ): return response.create_error_response( code=error.ERROR_CODE_ACCOUNT_WITHOUT_USER, message=error.ERROR_MESSAGE_ACCOUNT_WITHOUT_USER, status=400, ) # get vendor / subaccount info for user if set if ( self.orchard_user_id is not None and self.subaccount_id and self.vendor_id is None ) or ( self.orchard_user_id is not None and self.subaccount_id is None and self.vendor_id is None ): user_response = ows_users.get_user_details_by_user_id(self.orchard_user_id) if user_response.status >= 500: pass elif user_response.status != 200: return user_response elif user_response.message['type'] == 'alw': response_vendor_id = ( str(user_response.message['account'][context_constants.VENDOR_ID]) if user_response.message['account'][context_constants.VENDOR_ID] else None ) response_subaccount_id = ( str( user_response.message['account'][ context_constants.SUBACCOUNT_ID ] ) if user_response.message['account'][context_constants.SUBACCOUNT_ID] else None ) if self.vendor_id: if self.vendor_id != response_vendor_id: return response.create_error_response( code=error.ERROR_CODE_VENDOR_MISMATCH, message=error.ERROR_MESSAGE_VENDOR_MISMATCH, # noqa status=400, ) if self.subaccount_id: if self.subaccount_id != response_subaccount_id: return response.create_error_response( code=error.ERROR_CODE_SUBACCOUNT_MISMATCH, message=error.ERROR_MESSAGE_SUBACCOUNT_MISMATCH, # noqa status=400, ) self.vendor_id = response_vendor_id self.subaccount_id = response_subaccount_id else: self.vendor_id = None self.subaccount_id = None return True def to_dict(self): """Return a dict containing the full context.""" context = {} if self.identity_id: context[context_constants.IDENTITY_ID] = self.identity_id if self.identity_uuid: context[context_constants.IDENTITY_UUID] = self.identity_uuid if self.orchard_user_id: context[context_constants.USER_ID] = self.orchard_user_id # Add LabelProfile if self.orchard_user_id.startswith('alw:'): profile = self.orchard_user_id.replace( 'alw', headers.PROFILE_TYPE_LABEL ) context[context_constants.PROFILE_TYPE_AND_ID] = profile # Add OrchAdminProfile if self.orchard_user_id.startswith('oa:'): profile = self.orchard_user_id.replace( 'oa', headers.PROFILE_TYPE_ORCH_ADMIN ) context[context_constants.PROFILE_TYPE_AND_ID] = profile if self.vendor_id: context[context_constants.VENDOR_ID] = self.vendor_id if self.subaccount_id: context[context_constants.SUBACCOUNT_ID] = self.subaccount_id return context class ProfileContext: """Class to organize profile based context logic and validations.""" def __init__( self, profile_type=None, profile_id=None, identity_id=None, identity_uuid=None ): """Initialize class.""" self.profile_type = str(profile_type) if profile_type else None self.profile_id = str(profile_id) if profile_id else None self.identity_id = str(identity_id) if identity_id else None self.identity_uuid = str(identity_uuid) if identity_uuid else None def validate(self): """Validate initialization arguments.""" # profile type and profile id, no identity id if self.profile_id and self.profile_type and not self.identity_id: return True # just identity id if self.identity_id and not self.profile_id and not self.profile_type: return True # all 3 if self.identity_id and self.profile_id and self.profile_type: return True # none if not self.identity_id and not self.profile_id and not self.profile_type: return True self.profile_type = None self.profile_id = None self.identity_id = None return error_response.create_error_incomplete_profile_headers() def to_dict(self): """Return a dict containing the full context.""" context = {} if self.profile_type and self.profile_id: context[context_constants.PROFILE_TYPE_AND_ID] = '{}:{}'.format( self.profile_type, self.profile_id ) if self.identity_id: context[context_constants.IDENTITY_ID] = self.identity_id if self.identity_uuid: context[context_constants.IDENTITY_UUID] = self.identity_uuid # if LabelProfile, hyrdate alw user id & vendor/subaccount if self.profile_type == headers.PROFILE_TYPE_LABEL: legacy_alw_id = 'alw:{}'.format(self.profile_id) context[context_constants.USER_ID] = legacy_alw_id user_response = ows_users.get_user_details_by_user_id(legacy_alw_id) if user_response.status == 200: user_message = user_response.message['account'] vendor_id = user_message[context_constants.VENDOR_ID] subaccount_id = user_message[context_constants.SUBACCOUNT_ID] if vendor_id: context[context_constants.VENDOR_ID] = str(vendor_id) if subaccount_id: context[context_constants.SUBACCOUNT_ID] = str(subaccount_id) if user_response.status >= 500: pass # if OrchAdminProfile, hyrdate oa user id if self.profile_type == headers.PROFILE_TYPE_ORCH_ADMIN: context[context_constants.USER_ID] = 'oa:{}'.format(self.profile_id) return context