"""Account Resource Getters.""" from dataclasses import asdict from typing import Any from ddtrace import tracer from account.constants import constants, tenants from account.logic.resource_getters.attributes import Attributes, DistributorAttributes, Tenant from account.logic.vendor import lookup_vendors_by_uuids, lookup_vendors_by_vendor_ids class BaseAccountResourceGetter: """ Base class implementing common vendor lookup and tenant creation logic. Subclasses must implement: - lookup_vendor: to call the appropriate vendor lookup API. - build_attributes: to build the Attributes object. """ def __init__(self, vendor: str | int, fetch_flags: list[str]): # vendor_id or vendor_uuid self.vendor = vendor self.fetch_flags = fetch_flags @tracer.wrap() def get_attributes(self, *args: Any, **kwargs: Any) -> dict[str, Any]: # Get resource attributes about account for AuthorizationBackend. response = self.lookup_vendor([self.vendor], self.fetch_flags) if not response or not response.message: return {} vendors = response.message.get('vendors') if not isinstance(vendors, list) or not vendors: return {} account_data = vendors[0] if not account_data or 'uuid' not in account_data: return {} tenant = Tenant( tenant_uuid=account_data['uuid'], tenant_type=tenants.TenantType.ACCOUNT, tenant_hierarchy=[ account_data['company_brand_uuid'], account_data['parent_company_uuid'], ], ) attributes = self.build_attributes(tenant, account_data) return asdict(attributes) def lookup_vendor(self, vendor_identifiers: list[str | int], fetch_flags: list[str]) -> Any: raise NotImplementedError('Subclasses must implement lookup_vendor') def build_attributes(self, tenant: Tenant, account_data: dict[str, Any]) -> Attributes: raise NotImplementedError('Subclasses must implement build_attributes') class AccountByIdResourceGetter(BaseAccountResourceGetter): def __init__(self, vendor_id: int): super().__init__(vendor_id, [constants.FETCH_TENANT_HIERARCHY]) def lookup_vendor(self, vendor_identifiers: list[str | int], fetch_flags: list[str]) -> Any: return lookup_vendors_by_vendor_ids(vendor_identifiers, fetch_flags=fetch_flags) def build_attributes(self, tenant: Tenant, account_data: dict[str, Any]) -> Attributes: return Attributes( tenant=tenant, name=self.vendor, # PDP doesn't need this, but send it anyway for viz ) class AccountByUuidResourceGetter(BaseAccountResourceGetter): def __init__( self, vendor_uuid: str, ): super().__init__(vendor_uuid, [constants.FETCH_TENANT_HIERARCHY]) def lookup_vendor(self, vendor_identifiers: list[str | int], fetch_flags: list[str]) -> Any: return lookup_vendors_by_uuids(vendor_identifiers, fetch_flags=fetch_flags) def build_attributes(self, tenant: Tenant, account_data: dict[str, Any]) -> Attributes: return Attributes( tenant=tenant, name=self.vendor, # vendor_uuid used for name ) class AccountDistributorByUuidResourceGetter(BaseAccountResourceGetter): def __init__(self, vendor_uuid: str): super().__init__( vendor_uuid, [constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR] ) def lookup_vendor(self, vendor_identifiers: list[str | int], fetch_flags: list[str]) -> Any: return lookup_vendors_by_uuids(vendor_identifiers, fetch_flags=fetch_flags) def build_attributes( self, tenant: Tenant, account_data: dict[str, Any] ) -> DistributorAttributes: return DistributorAttributes( tenant=tenant, name=self.vendor, is_distributor=account_data['is_distributor'], ) class AccountDistributorByIdResourceGetter(BaseAccountResourceGetter): def __init__(self, vendor_id: int): super().__init__( vendor_id, [constants.FETCH_TENANT_HIERARCHY, constants.FETCH_IS_DISTRIBUTOR] ) def lookup_vendor(self, vendor_identifiers: list[str | int], fetch_flags: list[str]) -> Any: return lookup_vendors_by_vendor_ids(vendor_identifiers, fetch_flags=fetch_flags) def build_attributes( self, tenant: Tenant, account_data: dict[str, Any] ) -> DistributorAttributes: return DistributorAttributes( tenant=tenant, name=self.vendor, is_distributor=account_data['is_distributor'], )