Dynamic Resolver

dynamic_resolver.py

import asyncio
from graphql import (
    graphql, GraphQLSchema, GraphQLObjectType, GraphQLField, GraphQLString, GraphQLList, GraphQLInt,
    GraphQLArgument, GraphQLInputObjectType, GraphQLInputField, GraphQLNonNull
)

# Mock data for Accounts and Contacts
mock_data = {
    "Account": [
        {"Id": 1, "Name": "Edge Communications", "Industry": "Telecommunications"},
        {"Id": 2, "Name": "MediaCorp", "Industry": "Media"},
        {"Id": 3, "Name": "Tech Innovations", "Industry": "Technology"}
    ],
    "Contact": [
        {"Id": 1, "Name": "John Doe", "AccountId": 1},
        {"Id": 2, "Name": "Jane Smith", "AccountId": 2},
        {"Id": 3, "Name": "Emily White", "AccountId": 3}
    ]
}

def get_graphql_type(data_type):
    if data_type == "String":
        return GraphQLString
    elif data_type == "Integer":
        return GraphQLInt
    return GraphQLString

def create_edge_node_structure(name, node_type):
    edge_type = GraphQLObjectType(
        name=f"{name}Edge",
        fields={
            "node": GraphQLField(node_type)
        }
    )

    connection_type = GraphQLObjectType(
        name=f"{name}Connection",
        fields={
            "edges": GraphQLList(edge_type)
        }
    )
    return connection_type

entity_types = {}
filter_input_types = {}

database_schema = [
    {
        "entityName": "Account",
        "properties": {
            "Id": {"dataType": "Integer"},
            "Name": {"dataType": "String"},
            "Industry": {"dataType": "String"}
        },
        "relationships": {
            "hasMany": ["Contact"]
        }
    },
    {
        "entityName": "Contact",
        "properties": {
            "Id": {"dataType": "Integer"},
            "Name": {"dataType": "String"},
            "AccountId": {"dataType": "Integer"}
        },
        "relationships": {
            "belongsTo": ["Account"]
        }
    }
]

# Define GraphQL object types for each entity without relationships
for entity in database_schema:
    properties = entity["properties"]

    entity_fields = {
        prop_name: GraphQLField(get_graphql_type(prop_info["dataType"]), resolve=lambda obj, info, prop_name=prop_name: obj.get(prop_name))
        for prop_name, prop_info in properties.items()
    }

    entity_type = GraphQLObjectType(
        name=entity["entityName"], 
        fields=lambda entity_fields=entity_fields: entity_fields
    )
    entity_types[entity["entityName"]] = entity_type

    # Define the connection structure for each entity
    connection_type = create_edge_node_structure(entity["entityName"], entity_type)
    entity_types[f"{entity['entityName']}Connection"] = connection_type

# Add relationship fields dynamically
for entity in database_schema:
    relationships = entity.get("relationships", {})
    entity_fields = entity_types[entity["entityName"]].fields

    # Handle "hasMany" relationships
    for related_entity_name in relationships.get("hasMany", []):
        connection_type = entity_types[f"{related_entity_name}Connection"]

        def resolve_has_many(obj, info, related_entity_name=related_entity_name):
            return [{"node": item} for item in mock_data[related_entity_name] if item.get("AccountId") == obj["Id"]]

        entity_fields[related_entity_name] = GraphQLField(connection_type, resolve=resolve_has_many)

    # Update the entity type with new fields including relationships
    entity_types[entity["entityName"]] = GraphQLObjectType(name=entity["entityName"], fields=entity_fields)

# Define the where input types for filtering
for entity in database_schema:
    where_input_fields = {
        "Id": GraphQLInputField(GraphQLInt),
        "Name": GraphQLInputField(GraphQLString)
    }

    # Additional where input fields for related entities
    for related_entity in relationships.get("belongsTo", []):
        where_input_fields[related_entity] = GraphQLInputField(GraphQLInputObjectType(
            name=f"{related_entity}WhereInput",
            fields={
                "Industry": GraphQLInputField(GraphQLString)
            }
        ))

    filter_input_types[f"{entity['entityName']}WhereInput"] = GraphQLInputObjectType(
        name=f"{entity['entityName']}WhereInput",
        fields=where_input_fields
    )

# Define resolvers for root query fields
def resolve_entities(entity_name, info, where=None):
    filtered_data = mock_data[entity_name]
    if where:
        for key, value in where.items():
            if key in ["Id", "Name"]:
                filtered_data = [item for item in filtered_data if item.get(key) == value]
            # Handle nested filters for related entities
            elif key in entity_types and "Industry" in value:
                related_filtered_data = [item for item in mock_data[key] if item.get("Industry") == value["Industry"]]
                related_ids = [item["Id"] for item in related_filtered_data]
                filtered_data = [item for item in filtered_data if item.get(f"{key}Id") in related_ids]
    return [{"node": item} for item in filtered_data]

RootQuery and Schema definition
RootQuery = GraphQLObjectType(
    name='RootQueryType',
    fields={
        entity["entityName"]: GraphQLField(
             [f"{entity['entityName']}Connection"],
            args={
                'where': GraphQLArgument(filter_input_types[f"{entity['entityName']}WhereInput"])
            },
            resolve=lambda root, info, where=None, entity_name=entity["entityName"]: resolve_entities(entity_name, info, where)
        ) for entity in database_schema
    }
)

schema = GraphQLSchema(
    query=GraphQLObjectType(
        name='RootQueryType',
        fields={
            'hello': GraphQLField(
                GraphQLString,
                resolve=lambda obj, info: 'world')
        }))


schema = GraphQLSchema(query=RootQuery)

async def main():
    query = """
    {
      Contact(where: { Account: { Industry: { eq: "Media" } } }) {
        edges {
          node {
            Id
            Name
            Account {
              Name
            }
          }
        }
      }
    }
    """
    print('Running query...')
    result = await graphql(schema, query)
    print(result)

asyncio.run(main())