data_sources/connect_graphql.py

import requests
import os
import json
from utils import TEMP_DIR

def connect_graphql(graphql_url, api_token, graphql_token_header, session_hash):
    try:
        # Create the GraphQL Introspection Query
        query = """
                query IntrospectionQuery {
                    __schema {
                    queryType { name }
                    mutationType { name }
                    subscriptionType { name }
                    types {
                        ...FullType
                    }
                    directives {
                        name
                        description
                        locations
                        args {
                        ...InputValue
                        }
                    }
                    }
                }
                fragment FullType on __Type {
                    kind
                    name
                    description
                    fields(includeDeprecated: true) {
                    name
                    description
                    args {
                        ...InputValue
                    }
                    type {
                        ...TypeRef
                    }
                    isDeprecated
                    deprecationReason
                    }
                    inputFields {
                    ...InputValue
                    }
                    interfaces {
                    ...TypeRef
                    }
                    enumValues(includeDeprecated: true) {
                    name
                    description
                    isDeprecated
                    deprecationReason
                    }
                    possibleTypes {
                    ...TypeRef
                    }
                }
                fragment InputValue on __InputValue {
                    name
                    description
                    type { ...TypeRef }
                    defaultValue
                }
                fragment TypeRef on __Type {
                    kind
                    name
                    ofType {
                    kind
                    name
                    ofType {
                        kind
                        name
                        ofType {
                        kind
                        name
                        ofType {
                            kind
                            name
                            ofType {
                            kind
                            name
                            ofType {
                                kind
                                name
                                ofType {
                                kind
                                name
                                }
                            }
                            }
                        }
                        }
                    }
                    }
                }
                """
        print("Connecting to GraphQL Endpoint")
    
        # Access a database
        headers = {"Content-Type": "application/json"}
        if graphql_token_header and api_token:
            headers[graphql_token_header] = api_token
        response = requests.post(graphql_url, headers=headers, json={"query": query})
        response.raise_for_status()

        introspection_result = response.json()

        client_schema = introspection_result["data"]["__schema"]

        #Generate the list of types
        type_names_query = """
                query IntrospectionQuery {
                    __schema {
                        types {
                            name
                        }
                    }
                }
            """
        types_response = requests.post(graphql_url, headers=headers, json={"query": type_names_query})

        types_response_results =types_response.json()

        types_names = types_response_results["data"]

        type_names = []
        for name in types_names["__schema"]["types"]:
            type_names.append(name["name"])

        session_path = 'graphql'

        dir_path = TEMP_DIR / str(session_hash) / str(session_path)
        os.makedirs(dir_path, exist_ok=True)

        with open(f'{dir_path}/schema.json', 'w') as fp:
            json.dump(client_schema, fp, indent=2)

        return ["success","<p style='color:green;text-align:center;font-size:18px;'>GraphQL API connected successful</p>", type_names]
    except Exception as e:
        print("GraphQL CONNECTION ERROR")
        print(e)
        return ["error",f"<p style='color:red;text-align:center;font-size:18px;font-weight:bold;'>ERROR: {e}</p>"]