Router

• GraftRouter configures GraphQL queries and mutations automatically based on discovered models.

GraftRouter

Bases: GraphQLRouter

A custom router class that sets up a GraphQL API using schemas generated from SQLModel classes. It handles dynamic query and mutation generation, permissions, and context setup for each request.

Parameters:

Name	Type	Description	Default
query	object	A class containing GraphQL query resolvers.	None
mutation	object	A class containing GraphQL mutation resolvers.	None
context_getter	Callable	A function to get additional context for each request.	None
permission_getter	Callable	A function to determine permissions for operations.	None
dl_filter	Callable	A function to apply filters to data loaders.	None
query_filter	Callable	A function to apply filters to queries.	None
engine	Engine | Dict[str, Engine]	Database engine(s) used for SQL operations.	None
extensions	list	List of Strawberry extensions to be applied to the schema.	None
enable_queries	bool	Flag to enable query generation.	True
enable_put_mutations	bool	Flag to enable PUT mutations.	False
enable_delete_mutations	bool	Flag to enable DELETE mutations.	False
auto_foreign_keys	bool	Flag to automatically handle foreign keys.	False
**kwargs	Any	Additional keyword arguments passed to the base GraphQLRouter.	{}

Source code in graftql/core/router.py

class GraftRouter(GraphQLRouter):
    """
    A custom router class that sets up a GraphQL API using schemas generated from SQLModel classes.
    It handles dynamic query and mutation generation, permissions, and context setup for each request.

    Args:
        query (object): A class containing GraphQL query resolvers.
        mutation (object): A class containing GraphQL mutation resolvers.
        context_getter (Callable): A function to get additional context for each request.
        permission_getter (Callable): A function to determine permissions for operations.
        dl_filter (Callable): A function to apply filters to data loaders.
        query_filter (Callable): A function to apply filters to queries.
        engine (Engine | Dict[str, Engine]): Database engine(s) used for SQL operations.
        extensions (list): List of Strawberry extensions to be applied to the schema.
        enable_queries (bool): Flag to enable query generation.
        enable_put_mutations (bool): Flag to enable PUT mutations.
        enable_delete_mutations (bool): Flag to enable DELETE mutations.
        auto_foreign_keys (bool): Flag to automatically handle foreign keys.
        **kwargs (Any): Additional keyword arguments passed to the base GraphQLRouter.
    """

    functions = []

    def __init__(
        self,
        query: object | None = None,
        mutation: object | None = None,
        context_session: Callable | None = None,
        context_getter: Callable | None = None,
        permission_getter: Callable | None = None,
        dl_filter: Callable | None = None,
        query_filter: Callable | None = None,
        engine: AsyncEngine | Engine | dict[str, Engine] = None,
        extensions: list | None = None,
        enable_queries: bool = True,
        enable_put_mutations: bool = False,
        enable_delete_mutations: bool = False,
        auto_foreign_keys: bool = False,
        enable_federation_2: bool = False,
        **kwargs,
    ):
        if not query:
            query = genre_empty_query()
        if not mutation:
            mutation = genre_empty_mutation()
        if not extensions:
            extensions = []
        functions: dict[str, tuple] = {}
        Setup.setup(
            engine=engine,
            permission_getter=permission_getter,
            query_filter=query_filter,
        )
        need_query = True
        need_mutation = True
        for cls in Setup.classes.values():
            set_schema(cls, functions, auto_foreign_keys)
            if cls.__enable_query__ is None:
                cls.__enable_query__ = enable_queries
            if cls.__enable_put_mutation__ is None:
                cls.__enable_put_mutation__ = enable_put_mutations
            if cls.__enable_delete_mutation__ is None:
                cls.__enable_delete_mutation__ = enable_delete_mutations
            cls_query, cls_filter = get_query(cls)
            setattr(
                sys.modules[__name__],
                cls.__name__,
                cls.__strawberry_schema__,
            )
            setattr(
                sys.modules[__name__],
                cls.__name__ + "Filter",
                cls_filter,
            )
            if cls.__enable_query__:
                need_query = False

                def model_query_resolver() -> cls_query:
                    return cls_query()  # return an instance of the ModelQuery type

                # Create a Strawberry field for this query
                query_field = strawberry.field(description=f"Queries for {cls.__queryname__}")(
                    model_query_resolver
                )

                setattr(query, cls.__queryname__, query_field)

            cls_mutations = get_mutations(cls)
            # model_mutation_fields = {}
            if cls_mutations:
                # Create a resolver that returns an instance of ModelMutations
                def make_mutation_resolver(mutations):
                    def resolver() -> mutations:
                        return mutations()

                    return resolver

                resolver = make_mutation_resolver(cls_mutations)

                # Create a Strawberry field for this query
                mutation_field = strawberry.field(description=f"Mutations for {cls.__queryname__}")(
                    resolver
                )

                need_mutation = False
                setattr(mutation, cls.__queryname__, mutation_field)
        if need_query:
            query.hello_world = strawberry.field(hello_world)

        async def get_context(request: Request, response: Response) -> dict:
            context = {}
            if getattr(request, "state") and getattr(request.state, "db"):
                session = request.state.db
                context = {"engine": engine, "session": session}

            if context_getter:
                sig = inspect.signature(context_getter)
                if len(sig.parameters) >= 3:
                    context = await context_getter(request, response, context)
                else:
                    extra = await context_getter(request, response)
                    context.update(extra)
            for k, (func, return_class) in functions.items():
                context[k] = GraftLoader(
                    load_fn=(
                        func
                        if await Setup.permission_getter(return_class, context, "query")
                        else (
                            fake_dl_list
                            if type(inspect.signature(func).return_annotation) == GenericAlias
                            else fake_dl_one
                        )
                    ),
                    filter_method=dl_filter,
                    context=context,
                )
            return context

        if enable_federation_2:
            schema = strawberry.federation.Schema(
                query=strawberry.type(query),
                mutation=(
                    None
                    if need_mutation and hasattr(mutation, "__auto_generated__")
                    else strawberry.type(mutation)
                ),
                extensions=extensions,
                federation_version="2.11",
            )
        else:
            schema = strawberry.Schema(
                query=strawberry.type(query),
                mutation=(
                    None
                    if need_mutation and hasattr(mutation, "__auto_generated__")
                    else strawberry.type(mutation)
                ),
                extensions=extensions,
            )
        super().__init__(schema=schema, context_getter=get_context, **kwargs)

    async def process_result(
        self, request: Request, result: ExecutionResult
    ) -> GraphQLHTTPResponse:
        """
        Processes the GraphQL execution result, formatting errors and managing permissions.
        Args:
            request (Request): The incoming HTTP request.
            result (ExecutionResult): The result from executing a GraphQL operation.
        Returns:
            GraphQLHTTPResponse: The formatted response to be returned to the client.
        """
        data: GraphQLHTTPResponse = {"data": result.data}
        errors = []
        if "errors" in request.scope:
            errors.append(
                GraphQLError(
                    "User don't have necessary permissions for this path",
                    path=[c.__tablename__ for c in request.scope["errors"]],
                ).formatted
            )
        if result.errors:
            errors.extend([err.formatted for err in result.errors])
        if len(errors) > 0:
            data["errors"] = errors
        return data

process_result(request, result) async

Processes the GraphQL execution result, formatting errors and managing permissions. Args: request (Request): The incoming HTTP request. result (ExecutionResult): The result from executing a GraphQL operation. Returns: GraphQLHTTPResponse: The formatted response to be returned to the client.

Source code in graftql/core/router.py

async def process_result(
    self, request: Request, result: ExecutionResult
) -> GraphQLHTTPResponse:
    """
    Processes the GraphQL execution result, formatting errors and managing permissions.
    Args:
        request (Request): The incoming HTTP request.
        result (ExecutionResult): The result from executing a GraphQL operation.
    Returns:
        GraphQLHTTPResponse: The formatted response to be returned to the client.
    """
    data: GraphQLHTTPResponse = {"data": result.data}
    errors = []
    if "errors" in request.scope:
        errors.append(
            GraphQLError(
                "User don't have necessary permissions for this path",
                path=[c.__tablename__ for c in request.scope["errors"]],
            ).formatted
        )
    if result.errors:
        errors.extend([err.formatted for err in result.errors])
    if len(errors) > 0:
        data["errors"] = errors
    return data

fake_dl_list(keys) async

A placeholder asynchronous function that simulates fetching lists of items for each key. Args: keys (iterable): A collection of keys for which data needs to be fetched. Returns: An iterable of empty lists corresponding to each key.

Source code in graftql/core/router.py

async def fake_dl_list(keys):
    """
    A placeholder asynchronous function that simulates fetching lists of items for each key.
    Args:
        keys (iterable): A collection of keys for which data needs to be fetched.
    Returns:
        An iterable of empty lists corresponding to each key.
    """
    return {k: [] for k in keys}.values()

fake_dl_one(keys) async

A placeholder asynchronous function that simulates fetching single items for each key. Args: keys (iterable): A collection of keys for which data needs to be fetched. Returns: An iterable of None values corresponding to each key.

Source code in graftql/core/router.py

async def fake_dl_one(keys):
    """
    A placeholder asynchronous function that simulates fetching single items for each key.
    Args:
        keys (iterable): A collection of keys for which data needs to be fetched.
    Returns:
        An iterable of None values corresponding to each key.
    """
    return {k: None for k in keys}.values()

hello_world(self, info) async

A simple resolver function returning a greeting message. Returns: A string greeting 'Hello World'.

Source code in graftql/core/router.py

async def hello_world(self, info: strawberry.types.Info) -> str:
    """
    A simple resolver function returning a greeting message.
    Returns:
        A string greeting 'Hello World'.
    """
    return "Hello World"