"""GraphQL connector.""" import uuid import config from gql import Client from gql import gql as ggl_query from gql.transport.requests import RequestsHTTPTransport from src.constants.graphql import METADATA_QUERY from src.exceptions import GraphQLMetadataNotFound class GraphQLConnector: """graphql-router connector.""" def __init__( self, ): """Connector to graphql-router.""" auth_headers = { 'apollographql-client-name': config.SERVICE_NAME, 'Correlation-Id': str(uuid.uuid4()), 'Orchard-Identity-Id': config.USER_IDENTITY_ID, 'OrchardIdentity-Uuid': config.USER_IDENTITY_UUID, 'Orchard-Profile-Type': config.USER_PROFILE_TYPE, 'Orchard-Profile-Id': config.USER_PROFILE_ID, 'Orchard-Profile-Uuid': config.USER_PROFILE_UUID, 'Orchard-Roles': config.USER_ROLES, } env = config.QA_ENVIRONMENT \ if config.ENVIRONMENT == config.DEV_ENVIRONMENT \ else config.ENVIRONMENT url = f'https://{env}-graphql-router.theorchard.io/graphql' transport = RequestsHTTPTransport( url=url, use_json=True, headers=auth_headers, verify=True, ) self.client = Client( transport=transport, fetch_schema_from_transport=False, ) def make_request(self, query, params=None): """Make GraphQL request. Args: query (str): GraphQL query params (dict): query parameters Returns: GraphQL response object """ query = ggl_query(query) return self.client.execute(query, variable_values=params) def get_metadata_by_upc(self, upc): """Execute gql product metadata query. Args: upc (string): upc string for product Returns: dict[metadata] : list of product metadata """ params = {'upc': upc} result = {'metadata': {}} data = self.make_request(METADATA_QUERY, params=params) metadata = data['productByUpc'] if metadata is None: raise GraphQLMetadataNotFound( f'No metadata found for UPC: {upc}' ) result['metadata'] = metadata return result