GraphQL Integration¶

Cello provides first-class GraphQL support with decorator-based schema definition, DataLoader for N+1 prevention, and WebSocket-based subscriptions.

Quick Start¶

from cello import App
from cello.graphql import Query, Mutation, Schema, DataLoader, GraphQL

app = App()

@Query
def users(info) -> list:
    return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

@Query
def user(info, id: int) -> dict:
    return {"id": id, "name": "Alice"}

@Mutation
def create_user(info, name: str, email: str) -> dict:
    return {"id": 3, "name": name, "email": email}

schema = Schema().query(users).query(user).mutation(create_user).build()
result = await schema.execute("{ users }")

Decorators¶

@Query ¶

Marks a function as a GraphQL query resolver.

from cello.graphql import Query

@Query
def books(info) -> list:
    return db.get_all_books()

@Query
def book(info, id: int) -> dict:
    return db.get_book(id)

@Mutation ¶

Marks a function as a GraphQL mutation resolver.

from cello.graphql import Mutation

@Mutation
def create_book(info, title: str, author: str) -> dict:
    return db.create_book(title, author)

@Subscription ¶

Marks an async generator as a GraphQL subscription.

from cello.graphql import Subscription

@Subscription
async def book_added(info):
    async for event in event_stream("book_added"):
        yield event

DataLoader¶

Prevents N+1 query problems by batching and caching database calls.

from cello.graphql import DataLoader

async def batch_load_authors(ids):
    return [db.get_author(id) for id in ids]

author_loader = DataLoader(batch_fn=batch_load_authors)

# Single load (cached after first call)
author = await author_loader.load(1)

# Batch load
authors = await author_loader.load_many([1, 2, 3])

# Clear cache
author_loader.clear(1)       # Clear single key
author_loader.clear()         # Clear all

Schema Builder¶

Compose queries, mutations, and subscriptions into a schema using the fluent builder API.

from cello.graphql import Schema

schema = (
    Schema()
    .query(users)
    .query(user)
    .mutation(create_user)
    .mutation(update_user)
    .subscription(user_created)
    .build()
)

Class-Based Types¶

You can also register entire classes where each method becomes a resolver:

class QueryType:
    def users(self, info) -> list:
        return db.get_all_users()

    def books(self, info) -> list:
        return db.get_all_books()

schema = Schema().query(QueryType).build()

GraphQL Engine¶

Execute queries directly using the GraphQL engine.

from cello.graphql import GraphQL

engine = GraphQL()
engine.add_query(users)
engine.add_mutation(create_user)

result = await engine.execute("{ users }")
# {"data": {"users": [...]}, "errors": None}

# With variables
result = await engine.execute(
    "mutation($name: String!) { createUser(name: $name) }",
    variables={"name": "Alice"}
)

Configuration¶

Enable GraphQL on your Cello app:

from cello import App, GraphQLConfig

app = App()
app.enable_graphql(GraphQLConfig(
    path="/graphql",
    playground=True,
    introspection=True
))

Option	Default	Description
`path`	`/graphql`	GraphQL endpoint path
`playground`	`True`	Enable GraphiQL playground
`introspection`	`True`	Enable schema introspection

API Reference¶

Class	Description
`Query`	Decorator for query resolvers
`Mutation`	Decorator for mutation resolvers
`Subscription`	Decorator for subscription resolvers
`Field`	GraphQL field definition with optional resolver
`DataLoader`	Batching and caching loader for N+1 prevention
`GraphQL`	Execution engine for running queries
`Schema`	Fluent builder for composing a schema