Source code for biocontext_kb.utils._graphql

from gql import Client, gql  # type: ignore
from gql.transport.requests import RequestsHTTPTransport  # type: ignore
from graphql import GraphQLSchema




def fetch_graphql_schema(endpoint_url: str) -> GraphQLSchema:
    """Fetch the GraphQL schema from the given endpoint URL.

    Args:
        endpoint_url (str): The GraphQL endpoint URL.

    Returns:
        GraphQLSchema: The fetched GraphQL schema.
    """
    # Create a transport with your GraphQL endpoint
    transport = RequestsHTTPTransport(
        url=endpoint_url,
    )

    # Create a client
    client = Client(transport=transport, fetch_schema_from_transport=True)

    with client as _session:
        # The schema is automatically fetched and stored in the client
        if not client.schema:
            raise ValueError("Failed to fetch schema from the GraphQL endpoint.")

        return client.schema





def execute_graphql_query(endpoint_url, query_string, variables=None, headers=None) -> dict:
    """Make a generic GraphQL API call.

    Args:
        endpoint_url (str): The GraphQL endpoint URL
        query_string (str): The GraphQL query or mutation as a string
        variables (dict, optional): Variables for the GraphQL query
        headers (dict, optional): HTTP headers to include

    Returns:
        dict: The response data from the GraphQL API
    """
    # Set default headers if none provided
    if headers is None:
        headers = {
            "Content-Type": "application/json",
        }

    # Prepare the transport
    transport = RequestsHTTPTransport(
        url=endpoint_url,
        headers=headers,
        use_json=True,
    )

    # Create a client
    client = Client(transport=transport)

    # Parse the query string
    try:
        query = gql(query_string)
    except Exception as e:
        return {"status": "error", "message": f"Failed to parse query: {e!s}"}

    try:
        result = client.execute(query, variable_values=variables)
        return {"status": "success", "data": result}
    except Exception as e:
        return {"status": "error", "message": str(e)}