Skip to content

GraphQL API

ninja_gql

ninja-gql — GraphQL layer generator for Ninja Stack.

__all__ module-attribute 

__all__ = ['GqlGenerator', 'build_schema']

GqlGenerator 

GqlGenerator(schema: AgenticSchema)

Generates Strawberry types, input types, queries, and mutations from an ASD.

Source code in libs/ninja-gql/src/ninja_gql/generator.py 
53
54
55
56
57
58
59
60
def __init__(self, schema: AgenticSchema) -> None:
    self.schema = schema
    self._types: dict[str, type] = {}
    self._input_types: dict[str, type] = {}
    self._entity_map: dict[str, EntitySchema] = {e.name: e for e in schema.entities}
    self._rel_by_source: dict[str, list[RelationshipSchema]] = {}
    for rel in schema.relationships:
        self._rel_by_source.setdefault(rel.source_entity, []).append(rel)

schema instance-attribute 

schema = schema

generate_types 

generate_types() -> dict[str, type]

Create a Strawberry @strawberry.type class per entity.

Returns a mapping {EntityName: StrawberryType}.

Source code in libs/ninja-gql/src/ninja_gql/generator.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def generate_types(self) -> dict[str, type]:
    """Create a Strawberry ``@strawberry.type`` class per entity.

    Returns a mapping ``{EntityName: StrawberryType}``.
    """
    if self._types:
        return self._types

    # First pass: create basic types (no relationships yet)
    for entity in self.schema.entities:
        self._types[entity.name] = self._make_type(entity)

    # Second pass: wire relationship fields
    for entity in self.schema.entities:
        self._attach_relationships(entity)

    return self._types

generate_input_types 

generate_input_types() -> dict[str, tuple[type, type]]

Create Strawberry input types for create and update operations.

Returns {EntityName: (CreateInput, UpdateInput)}.

Source code in libs/ninja-gql/src/ninja_gql/generator.py 
82
83
84
85
86
87
88
89
90
91
92
93
def generate_input_types(self) -> dict[str, tuple[type, type]]:
    """Create Strawberry input types for create and update operations.

    Returns ``{EntityName: (CreateInput, UpdateInput)}``.
    """
    if self._input_types:
        return self._input_types

    for entity in self.schema.entities:
        create_cls, update_cls = self._make_input_types(entity)
        self._input_types[entity.name] = (create_cls, update_cls)
    return self._input_types

get_type 

get_type(entity_name: str) -> type

Return the generated Strawberry type for entity_name.

Source code in libs/ninja-gql/src/ninja_gql/generator.py 
95
96
97
98
99
def get_type(self, entity_name: str) -> type:
    """Return the generated Strawberry type for *entity_name*."""
    if not self._types:
        self.generate_types()
    return self._types[entity_name]

has_embeddable_fields 

has_embeddable_fields(entity: EntitySchema) -> bool

Return True if the entity has any field with embedding config.

Source code in libs/ninja-gql/src/ninja_gql/generator.py 
101
102
103
def has_embeddable_fields(self, entity: EntitySchema) -> bool:
    """Return True if the entity has any field with embedding config."""
    return any(f.embedding is not None for f in entity.fields)

build_schema 

build_schema(
    asd: AgenticSchema,
    repo_getter: Callable[[str], Repository[Any]]
    | None = None,
    agent_router: AgentRouter | None = None,
) -> Schema

Build a Strawberry Schema from an Agentic Schema Definition.

Parameters

asd: The project's ASD. repo_getter: A callable (entity_name) -> Repository. When None a stub that raises at call-time is used (useful for schema-only validation). agent_router: Optional agent router for ask_* queries.

Source code in libs/ninja-gql/src/ninja_gql/schema.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def build_schema(
    asd: AgenticSchema,
    repo_getter: Callable[[str], Repository[Any]] | None = None,
    agent_router: AgentRouter | None = None,
) -> strawberry.Schema:
    """Build a Strawberry ``Schema`` from an Agentic Schema Definition.

    Parameters
    ----------
    asd:
        The project's ASD.
    repo_getter:
        A callable ``(entity_name) -> Repository``.  When ``None`` a stub
        that raises at call-time is used (useful for schema-only validation).
    agent_router:
        Optional agent router for ``ask_*`` queries.
    """
    gen = GqlGenerator(asd)
    types = gen.generate_types()
    gen.generate_input_types()

    if repo_getter is None:

        def _no_repo(name: str) -> Repository[Any]:
            raise RuntimeError(f"No repository configured for {name}")

        repo_getter = _no_repo

    # -- build Query fields --------------------------------------------------
    query_annotations: dict[str, Any] = {}
    query_ns: dict[str, Any] = {"__annotations__": query_annotations}

    for entity in asd.entities:
        gql_type = types[entity.name]
        snake = _snake(entity.name)

        # get
        get_name = f"get_{snake}"
        get_fn = make_get_resolver(entity, gql_type, repo_getter)
        query_ns[get_name] = strawberry.field(resolver=get_fn)

        # list
        list_name = f"list_{snake}"
        list_fn = make_list_resolver(entity, gql_type, repo_getter)
        query_ns[list_name] = strawberry.field(resolver=list_fn)

        # semantic search (only for entities with embeddable fields)
        if gen.has_embeddable_fields(entity):
            search_name = f"search_{snake}"
            search_fn = make_search_resolver(entity, gql_type, repo_getter)
            query_ns[search_name] = strawberry.field(resolver=search_fn)

    # agent queries per domain
    for domain in asd.domains:
        ask_name = f"ask_{domain.name.lower().replace(' ', '_')}"
        ask_fn = make_agent_query_resolver(domain.name, agent_router)
        query_ns[ask_name] = strawberry.field(resolver=ask_fn)

    query_cls = type("Query", (), query_ns)
    Query = strawberry.type(query_cls, description="Auto-generated root Query")

    # -- build Mutation fields -----------------------------------------------
    mutation_annotations: dict[str, Any] = {}
    mutation_ns: dict[str, Any] = {"__annotations__": mutation_annotations}

    for entity in asd.entities:
        gql_type = types[entity.name]
        snake = _snake(entity.name)

        create_fn = make_create_resolver(entity, gql_type, repo_getter)
        mutation_ns[f"create_{snake}"] = strawberry.mutation(resolver=create_fn)

        update_fn = make_update_resolver(entity, gql_type, repo_getter)
        mutation_ns[f"update_{snake}"] = strawberry.mutation(resolver=update_fn)

        delete_fn = make_delete_resolver(entity, repo_getter)
        mutation_ns[f"delete_{snake}"] = strawberry.mutation(resolver=delete_fn)

    mutation_cls = type("Mutation", (), mutation_ns)
    Mutation = strawberry.type(mutation_cls, description="Auto-generated root Mutation")

    return strawberry.Schema(query=Query, mutation=Mutation)