API Reference

Protocols

PersistedQueryStore

Store for persisted query mappings.

Implement this protocol to provide custom storage for persisted queries.

async def get(hash: str) -> str | None

Get a query by its hash.

Parameters

Parameter	Type	Description
`hash`	str	SHA256 hash of the query.

Returns

Type	Description
str | None	The query string if found, None otherwise.

async def put(
    hash: str,
    query: str
) -> None

Store a query with its hash.

Parameters

Parameter	Type	Description
`hash`	str	SHA256 hash of the query.
`query`	str	The full query string.

---

Classes

APQHandler

Handler for Automatic Persisted Queries.

Implements the APQ protocol:

1. Client sends query hash → Server looks up full query
2. If not found, client sends hash + full query → Server stores
3. Subsequent requests use just the hash

def __init__(
    store: PersistedQueryStore,
    enabled: bool = True
)

Initialize the APQ handler.

Parameters

Parameter	Type	Description
`store`	PersistedQueryStore	Persisted query store implementation.
`enabled`	bool	Whether APQ is enabled.

property enabled() -> bool

Check if APQ is enabled.

async def resolve_query(
    query: str | None,
    extensions: dict | None = None
) -> APQResult

Resolve the full query using APQ.

Parameters

Parameter	Type	Description
`query`	str | None	The query from the request (may be None for hash-only).
`extensions`	dict | None	The extensions from the GraphQL request.

Returns

Type	Description
APQResult	APQResult with the resolved query.

def create_extension_response(hash: str) -> dict

Create the APQ extension response.

Parameters

Parameter	Type	Description
`hash`	str	The query hash.

Returns

Type	Description
dict	Extension dict for the GraphQL response.

---

APQResult

Result of APQ lookup.

Attributes: query: The full query string (if found). is_persisted: Whether the query was found in the store. hash: The hash used for lookup.

---

AbstractPermission

Base class for GraphQL permissions.

Permission classes control access to GraphQL fields. They are checked before field resolution and can raise exceptions or return False to deny access.

Attributes: message: Error message when permission is denied.

Example

class IsAuthenticated(AbstractPermission):
    message = "Authentication required"

    async def has_permission(self, source, info, **kwargs):
        return info.context.user is not None

@strawberry.type
class User:
    @field(permission_classes=[IsAuthenticated])
    async def email(self) -> str:
        return self._email

async def has_permission(
    source: Any,
    info: Info,
    **kwargs: Any
) -> bool

Check if the permission is granted.

Parameters

Parameter	Type	Description
`source`	Any	The source object (parent resolver result).
`info`	Info	GraphQL resolver info containing context. **kwargs: Additional arguments passed to the field.

Returns

Type	Description
bool	True if permission is granted, False otherwise.

Raises

Exception	Description
Exception	If permission is denied with a custom error.

---

AfterExecuteEvent

Emitted after a GraphQL operation completes successfully.

Contains the execution context with timing information and the raw response.

---

AliasLimitValidator

Validate query alias count against a limit.

Analyzes GraphQL queries to count aliases and validates against a configured maximum.

Example

validator = AliasLimitValidator(max_aliases=10)

count = validator.count_aliases(document)
validator.validate(document)  # Raises if too many

def __init__(max_aliases: int = 10) -> None

Initialize the validator.

Parameters

Parameter	Type	Description
`max_aliases`	int	Maximum allowed aliases.

property max_aliases() -> int

Get maximum alias limit.

def count_aliases(document: DocumentNode) -> int

Count the number of aliases in a query.

Parameters

Parameter	Type	Description
`document`	DocumentNode	Parsed GraphQL document.

Returns

Type	Description
int	Number of aliases.

def validate(document: DocumentNode) -> None

Validate alias count.

Parameters

Parameter	Type	Description
`document`	DocumentNode	Parsed GraphQL document.

Raises

Exception	Description
SecurityError	If query has too many aliases.

---

AllowAny

Permission that allows any access (no restrictions).

async def has_permission(
    source: Any,
    info: Info[Any, Any],
    **kwargs: Any
) -> bool

Always allow access.

---

BeforeExecuteEvent

Emitted immediately before a GraphQL operation is sent to Strawberry.

Subscribers may inspect the execution context (query, variables, user) but must NOT modify it — the event is immutable by design.

---

CacheBackendPersistedQueryStore

CacheBackendProtocol-backed APQ store for multi-process deployments.

Uses the platform’s CacheBackendProtocol abstraction rather than a raw Redis client — compatible with any configured cache provider (Redis, Memcached, etc.) without coupling this package to a specific driver.

Suitable for production use with multiple application instances.

def __init__(
    cache: Any,
    key_prefix: str = 'graphql:apq:',
    ttl_seconds: int = 86400
) -> None

Initialize the store.

Parameters

Parameter	Type	Description
`cache`	Any	A CacheBackendProtocol instance.
`key_prefix`	str	Prefix applied to all cache keys.
`ttl_seconds`	int	Time-to-live for stored entries in seconds.

async def get(hash: str) -> str | None

Get a query by its hash.

async def put(
    hash: str,
    query: str
) -> None

Store a query with its hash.

---

CacheConfig

Cache configuration for GraphQL responses.

Attributes: enabled: Whether caching is enabled. default_max_age: Default cache duration in seconds. default_scope: Default cache scope (PUBLIC/PRIVATE). vary_headers: Headers to vary cache on.

---

CacheControl

Cache control settings for a field or type.

def to_header() -> str

Generate Cache-Control header value.

---

CacheScope

Cache control scope for responses.

---

Connection

GraphQL Connection type template for cursor-based pagination.

This is a generic template. Use create_connection_type() to create concrete Connection types for your node types.

Attributes: edges: List of edges containing nodes. page_info: Pagination information. total_count: Total number of items.

---

ContextFactory

Factory for creating GraphQL contexts.

Provides a standardized way to create execution contexts with proper initialization and DataLoaderProtocol setup.

def __init__(
    config: GraphQLConfig | None = None,
    dataloader_factories: dict[str, Any] | None = None,
    enable_identity_resolution: bool | None = None,
    resolver: Any | None = None
) -> None

Initialize the context factory.

Parameters

Parameter	Type	Description
`config`	GraphQLConfig | None	GraphQL configuration.
`dataloader_factories`	dict[str, Any] | None	Factory functions for DataLoaders.
`enable_identity_resolution`	bool | None	Whether to automatically resolve OAuth IDs to UUIDs. If not provided, reads from config if available, otherwise defaults to False (opt-in).
`resolver`	Any | None	Optional DI resolver to use for identity resolution.

async def create_context(
    request: GraphQLRequest | None = None,
    user: Any | None = None,
    metadata: dict[str, Any] | None = None,
    raw_request: Any | None = None
) -> GraphQLContext

Create a new GraphQL context.

This async version properly resolves OAuth external IDs to internal UUIDs if the OAuthIdentityStore is available in the container, and optionally authenticates and resolves a principal.

Authentication flow:

1. If user is provided, use it directly (middleware-authenticated)
2. If no user but raw_request is available, optionally authenticate
3. Resolve principal via GraphQLPrincipalResolverProtocol
4. If no resolver exists but user is present, create fallback principal

Parameters

Parameter	Type	Description
`request`	GraphQLRequest | None	The GraphQL request.
`user`	Any | None	Current user (from middleware or previous auth).
`metadata`	dict[str, Any] | None	Additional metadata.
`raw_request`	Any | None	The raw HTTP request.

Returns

Type	Description
GraphQLContext	A new GraphQLContext instance with resolved user ID and principal.

def create_context_sync(
    request: GraphQLRequest | None = None,
    user: Any | None = None,
    metadata: dict[str, Any] | None = None,
    raw_request: Any | None = None
) -> GraphQLContext

Create a GraphQL context synchronously (testing/sync-bridge only).

Warning Prefer create_context in production code. This method does not perform async identity resolution or principal resolution — the user object is passed through as-is and principal is set to None. Use only in test environments or sync integration points that cannot await.

Parameters

Parameter	Type	Description
`request`	GraphQLRequest | None	The GraphQL request.
`user`	Any | None	Current user (no async identity resolution performed).
`metadata`	dict[str, Any] | None	Additional metadata.
`raw_request`	Any | None	The raw HTTP request.

Returns

Type	Description
GraphQLContext	A new GraphQLContext instance without async identity or principal resolution.

def register_dataloader(
    name: str,
    factory: Any
) -> None

Register a DataLoaderProtocol factory.

Parameters

Parameter	Type	Description
`name`	str	DataLoaderProtocol name.
`factory`	Any	Factory function that creates the DataLoaderProtocol.

def from_dict(
    data: dict[str, Any],
    request: GraphQLRequest | None = None
) -> GraphQLContext

Create a GraphQLContext from a plain dictionary.

A convenience factory for callers that previously passed dict directly to GraphQLExecutorProtocol.execute. Extracts user, metadata, and optionally request from data.

Parameters

Parameter	Type	Description
`data`	dict[str, Any]	Mapping with optional keys ``user``, ``metadata``, and ``request``.
`request`	GraphQLRequest | None	Explicit request object; overrides ``data["request"]`` when provided.

Returns

Type	Description
GraphQLContext	A fully initialised GraphQLContext.

Example

context = ContextFactory.from_dict(
    {"user": current_user, "metadata": {"source": "api"}},
)
result = await executor.execute(query, context=context)

---

CursorConnection

A connection for cursor-based pagination.

---

CursorPaginationInput

Input for cursor-based pagination.

Attributes: first: Number of items to fetch from the start. after: Cursor to fetch items after. last: Number of items to fetch from the end. before: Cursor to fetch items before.

---

DataLoaderConfig

DataLoaderProtocol configuration.

Attributes: enabled: Whether DataLoaderProtocol integration is enabled. batch_enabled: Whether batching is enabled. cache_enabled: Whether per-request caching is enabled. max_batch_size: Maximum batch size for DataLoaderProtocol. batch_schedule_fn: How to schedule batch execution.

---

DataLoaderProtocol

DataLoaderProtocol for batching and caching.

DataLoaderProtocol is a utility for batching and caching data fetches to efficiently resolve GraphQL queries and avoid the N+1 problem.

Example

async def batch_load_users(ids: list[str]) -> list[User | None]:
    users = await db.get_users_by_ids(ids)
    # Return in same order as input ids
    user_map = {u.id: u for u in users}
    return list(map(lambda id: user_map.get(id), ids))

loader = DataLoaderProtocol(batch_load_users)

# These will be batched into a single call
user1 = await loader.load("1")
user2 = await loader.load("2")

def __init__(
    batch_fn: Callable[[list[K]], Awaitable[list[V | None]]],
    config: DataLoaderConfig | None = None,
    cache: LoaderCache[K, V] | None = None
) -> None

Initialize the DataLoaderProtocol.

Parameters

Parameter	Type	Description
`batch_fn`	Callable[[list[K]], Awaitable[list[V | None]]]	Function that loads a batch of values by keys.
`config`	DataLoaderConfig | None	DataLoaderProtocol configuration.
`cache`	LoaderCache[K, V] | None	Optional cache implementation.

property config() -> DataLoaderConfig

Get configuration.

async def load(key: K) -> V | None

Load a value by key.

The load will be batched with other loads that occur in the same tick of the event loop.

Parameters

Parameter	Type	Description
`key`	K	The key to load.

Returns

Type	Description
V | None	The loaded value or None if not found.

Raises

Exception	Description
DataLoaderError	If loading fails.

async def load_many(keys: Sequence[K]) -> list[V | None]

Load multiple values by keys.

Parameters

Parameter	Type	Description
`keys`	Sequence[K]	List of keys to load.

Returns

Type	Description
list[V | None]	List of values in the same order as keys.

def prime(
    key: K,
    value: V
) -> None

Prime the cache with a value.

Use this to add values to the cache that were loaded through other means.

Parameters

Parameter	Type	Description
`key`	K	The key.
`value`	V	The value to cache.

def clear(key: K) -> None

Clear a cached value.

Parameters

Parameter	Type	Description
`key`	K	The key to clear.

def clear_all() -> None

Clear all cached values.

---

DataLoaderStats

Statistics for a DataLoaderProtocol instance.

property cache_hit_ratio() -> float

Calculate cache hit ratio.

---

DeleteResult

Result of a delete mutation.

Attributes: success: Whether deletion succeeded. id: ID of deleted item. message: Optional message.

---

DenyAll

Permission that denies all access.

async def has_permission(
    source: Any,
    info: Info[Any, Any],
    **kwargs: Any
) -> bool

Always deny access.

---

DeprecationDirectiveHandler

Marks targets as deprecated using the ``@deprecated`` directive.

Adds a __deprecated__ attribute with the deprecation reason to any target object that supports attribute assignment.

def apply_directive(
    directive_name: str,
    args: dict[str, Any],
    target: Any
) -> Any

Apply @deprecated by setting __deprecated__ on target.

Parameters

Parameter	Type	Description
`directive_name`	str	Expected to be ``"deprecated"``.
`args`	dict[str, Any]	May contain ``"reason"`` (str).
`target`	Any	Any object.

Returns

Type	Description
Any	The *target* with ``__deprecated__`` set.

---

DepthLimitConfig

Query depth limiting configuration.

Attributes: enabled: Whether depth limiting is enabled. max_depth: Maximum allowed query depth. ignore_introspection: Ignore introspection queries.

---

DepthLimitExtension

Strawberry extension for query depth limiting.

Add this extension to your schema to automatically validate query depth before execution.

Example

from lexigram.graphql.security import DepthLimitExtension

schema = strawberry.Schema(
    query=Query,
    extensions=[DepthLimitExtension(max_depth=10)],
)

def __init__(
    max_depth: int = 10,
    ignore_introspection: bool = True
) -> None

Initialize the extension.

Parameters

Parameter	Type	Description
`max_depth`	int	Maximum allowed query depth.
`ignore_introspection`	bool	Skip introspection queries.

def on_operation() -> Iterator[None]

Hook called during operation execution.

---

DepthLimitValidator

Validate query depth against a limit.

Analyzes GraphQL queries to determine their depth and validates against a configured maximum depth.

Example

validator = DepthLimitValidator(max_depth=10)

# Validate a parsed document
depth = validator.get_depth(document)
validator.validate(document)  # Raises if too deep

def __init__(
    max_depth: int = 10,
    ignore_introspection: bool = True
) -> None

Initialize the validator.

Parameters

Parameter	Type	Description
`max_depth`	int	Maximum allowed query depth.
`ignore_introspection`	bool	Skip depth check for introspection queries.

property max_depth() -> int

Get maximum depth limit.

def get_depth(document: DocumentNode) -> int

Calculate the maximum depth of a query.

Parameters

Parameter	Type	Description
`document`	DocumentNode	Parsed GraphQL document.

Returns

Type	Description
int	Maximum depth of the query.

def validate(document: DocumentNode) -> None

Validate query depth.

Parameters

Parameter	Type	Description
`document`	DocumentNode	Parsed GraphQL document.

Raises

Exception	Description
DepthLimitError	If query exceeds depth limit.

---

DirectiveLocation

GraphQL directive locations.

---

DirectiveRegistry

Registry-based implementation of DirectiveHandler.

Maps directive names to handler callables via on (decorator syntax or explicit register call). When apply_directive is invoked with an unknown directive name, the target is returned unchanged if no default handler is registered, or the default handler is called.

Parameters

Parameter	Type	Description
`default_handler`		Optional fallback for unrecognised directives. Receives the same ``(directive_name, args, target)`` signature. When ``None`` (default), unknown directives are silently ignored.

def __init__(default_handler: Any | None = None) -> None

def register(
    directive_name: str,
    handler: Any
) -> None

Register a handler callable for a directive.

Parameters

Parameter	Type	Description
`directive_name`	str	Name of the GraphQL directive (without ``@``).
`handler`	Any	Callable ``(directive_name, args, target) -> target``.

def on(directive_name: str) -> Any

Decorator that registers a handler for directive_name.

Example

@registry.on("auth")
def apply_auth(name, args, target):
    target.__roles__ = args.get("roles", [])
    return target

Parameters

Parameter	Type	Description
`directive_name`	str	Name of the directive.

Returns

Type	Description
Any	Decorator that registers the decorated function.

def apply_directive(
    directive_name: str,
    args: dict[str, Any],
    target: Any
) -> Any

Apply a registered directive handler to target.

If no handler is registered under directive_name the default handler is called if one was supplied; otherwise target is returned as-is.

Parameters

Parameter	Type	Description
`directive_name`	str	Directive to apply (without ``@``).
`args`	dict[str, Any]	Directive arguments from the schema.
`target`	Any	Schema element to transform.

Returns

Type	Description
Any	The (possibly transformed) *target*.

---

Edge

An edge in a connection.

---

ErrorConfig

Error handling configuration.

Attributes: mask_errors: Whether to mask internal errors in production. include_stacktrace: Whether to include stacktrace in errors. debug_mode: Whether debug mode is enabled. log_errors: Whether to log errors.

---

ExecutionContextProtocol

Context for a single query execution.

Tracks execution state, metrics, and provides hooks for middleware and extensions.

Attributes: context: The GraphQL context. operation_type: Type of operation being executed. start_time: Execution start time. end_time: Execution end time. errors: Errors encountered during execution. extensions: Execution extensions data.

def mark_complete() -> None

Mark execution as complete.

property duration_ms() -> float

Get execution duration in milliseconds.

def add_error(error: Exception) -> None

Add an error to the execution context.

property has_errors() -> bool

Check if execution has errors.

---

ExecutionTrace

Complete execution trace.

Attributes: operation_name: Name of the operation. start_time: Execution start time. end_time: Execution end time. total_duration_ms: Total duration. root_span: Root span of the trace. resolver_count: Number of resolvers called.

def end() -> None

End the trace.

def to_dict() -> dict[str, Any]

Convert to dictionary.

Returns

Type	Description
dict[str, Any]	Dictionary representation (Apollo tracing format).

---

FieldInfo

Information about a GraphQL field.

---

GraphQLConfig

Hierarchical root configuration for Lexigram GraphQL.

Attributes: name: Configuration name (default: “graphql”) enabled: Whether GraphQL module is enabled path: GraphQL endpoint path debug: Platform debug flag (aligns with ObservabilityConfig.debug). Propagates to errors.debug_mode. enable_identity_resolution: Resolve OAuth IDs to internal UUIDs cache: Response caching settings depth_limit: Query depth limiting complexity: Query complexity settings persisted_queries: Persisted queries settings batch: Batch query settings introspection: Introspection settings playground: GraphQL playground settings subscriptions: WebSocket subscription settings dataloader: DataLoaderProtocol settings tracing: Tracing settings metrics: Metrics settings error: Error handling settings rate_limit: Rate limiting settings

def development(cls) -> GraphQLConfig

Create development configuration with debugging enabled.

def production(cls) -> GraphQLConfig

Create production configuration with security hardening.

---

GraphQLContext

GraphQL execution context.

Provides context for GraphQL operations including request information, user data, and configuration.

Attributes: request_id: Unique request identifier. user: Current user (if authenticated). principal: Resolved principal for identity access across resolvers. request: The GraphQL request. config: GraphQL configuration. started_at: Request start time. metadata: Additional context metadata. dataloaders: DataLoaderProtocol instances by name.

def get_dataloader(name: str) -> Any | None

Get a DataLoaderProtocol by name.

Parameters

Parameter	Type	Description
`name`	str	DataLoaderProtocol name.

Returns

Type	Description
Any | None	The DataLoaderProtocol instance or None.

def set_dataloader(
    name: str,
    loader: Any
) -> None

Set a DataLoaderProtocol.

Parameters

Parameter	Type	Description
`name`	str	DataLoaderProtocol name.
`loader`	Any	DataLoaderProtocol instance.

def get_metadata(
    key: str,
    default: Any = None
) -> Any

Get metadata value.

Parameters

Parameter	Type	Description
`key`	str	Metadata key.
`default`	Any	Default value if not found.

Returns

Type	Description
Any	The metadata value.

def set_metadata(
    key: str,
    value: Any
) -> None

Set metadata value.

Parameters

Parameter	Type	Description
`key`	str	Metadata key.
`value`	Any	Metadata value.

property operation_name() -> str | None

Get the operation name from request.

property variables() -> dict[str, Any]

Get variables from request.

property elapsed_ms() -> float

Get elapsed time in milliseconds.

async def dispose_scope() -> None

Dispose the per-request DI scope, releasing all scoped services.

No-op when no scope was created for this context. The GraphQL executor calls this automatically; application code should not normally need to call it directly.

---

GraphQLController

GraphQL HTTP endpoint controller.

This controller exposes GraphQL queries and mutations via HTTP POST at the configured path (default: /graphql).

Example

Registration is handled automatically by the GraphQL provider. To mount the controller manually, resolve the web application via the HTTPApplicationProtocol from contracts — do not import from lexigram.web directly

from lexigram.contracts.web.protocols import HTTPApplicationProtocol
# app = await container.resolve(HTTPApplicationProtocol)
# app.mount("/graphql", graphql_view)

def __init__(provider: GraphQLProvider | None = None) -> None

Initialize the GraphQL controller.

Parameters

Parameter	Type	Description
`provider`	GraphQLProvider | None	GraphQL provider instance. If None, will be resolved from the DI container at request time.

def collect_routes(cls) -> list[dict[str, Any]]

Collect routes from controller methods.

async def execute(request: GraphQLRequestProtocol) -> JSONResponse

Execute a GraphQL query.

Parameters

Parameter	Type	Description
`request`	GraphQLRequestProtocol	HTTP request containing the GraphQL query.

Returns

Type	Description
JSONResponse	JSON response with query results or errors.

---

GraphQLErrorCode

Standard GraphQL error codes.

---

GraphQLErrorData

Structured GraphQL error data.

def to_dict() -> dict[str, Any]

Convert to standard GraphQL error format.

---

GraphQLErrorExtensions

Extended error information.

def to_dict() -> dict[str, Any]

Convert to dictionary for JSON serialization.

---

GraphQLExecutorProtocol

GraphQL query executor.

Provides a high-level interface for executing GraphQL operations with proper error handling, timeout support, and metrics collection.

Example

from strawberry import Schema
from lexigram.graphql.core import GraphQLExecutorProtocol

schema = Schema(query=Query)
executor = GraphQLExecutorProtocol(schema)

response = await executor.execute(
    query="{ hello }",
    variables={},
)

def __init__(
    schema: StrawberrySchema,
    timeout_secs: float | None = None,
    middleware: list[Callable[Ellipsis, Any]] | None = None,
    error_config: ErrorConfig | None = None,
    event_bus: EventBusProtocol | None = None
) -> None

Initialize the executor.

Parameters

Parameter	Type	Description
`schema`	StrawberrySchema	Strawberry GraphQL schema.
`timeout_secs`	float | None	Default timeout in seconds.
`middleware`	list[Callable[Ellipsis, Any]] | None	List of middleware functions.
`error_config`	ErrorConfig | None	Error configuration for formatting.
`event_bus`	EventBusProtocol | None	Optional EventBusProtocol for lifecycle event publishing. Subscribers receive BeforeExecuteEvent, AfterExecuteEvent, and OnErrorEvent.

property schema() -> StrawberrySchema

Get the GraphQL schema.

async def execute(
    query: str,
    variables: dict[str, Any] | None = None,
    operation_name: str | None = None,
    context: GraphQLContext | None = None,
    timeout_secs: float | None = None
) -> Result[GraphQLResponse[Any], Exception]

Execute a GraphQL query.

Parameters

Parameter	Type	Description
`query`	str	GraphQL query string.
`variables`	dict[str, Any] | None	Query variables.
`operation_name`	str | None	Operation name to execute.
`context`	GraphQLContext | None	GraphQL context. Use ContextFactory.from_dict to convert a plain dictionary before passing it here.
`timeout`		Timeout in seconds (overrides default).

Returns

Type	Description
Result[GraphQLResponse, Exception]	Result wrapping the response or an infrastructure error.

Raises

Exception	Description
None	Infrastructure errors are wrapped in Err.

def get_metrics(execution_context: ExecutionContextProtocol) -> QueryMetrics

Get execution metrics.

Parameters

Parameter	Type	Description
`execution_context`	ExecutionContextProtocol	Execution context.

Returns

Type	Description
QueryMetrics	Query metrics.

---

GraphQLLocation

Location in a GraphQL document.

---

GraphQLMetrics

Aggregated GraphQL metrics.

Attributes: total_requests: Total number of requests. successful_requests: Number of successful requests. failed_requests: Number of failed requests. total_duration_ms: Total execution time. avg_duration_ms: Average execution time. operations_by_type: Count by operation type. operations_by_name: Count by operation name. errors_by_type: Error count by type.

def record_request(stats: QueryStats) -> None

Record a request’s statistics.

Parameters

Parameter	Type	Description
`stats`	QueryStats	Query statistics.

def record_error(error_type: str) -> None

Record an error.

Parameters

Parameter	Type	Description
`error_type`	str	Type of error.

def to_dict() -> dict[str, Any]

Convert to dictionary.

Returns

Type	Description
dict[str, Any]	Dictionary representation.

---

GraphQLModule

GraphQL layer (Strawberry): schema building, execution, federation, and subscriptions.

Call configure to configure and mount the GraphQL endpoint.

Usage

import strawberry

@strawberry.type
class Query:
    @strawberry.field
    def hello(self) -> str:
        return "world"

@module(
    imports=[GraphQLModule.configure(query_class=Query)]
)
class AppModule(Module):
    pass

def configure(
    cls,
    config: Any | None = None,
    query_class: Any | None = None,
    mutation_class: Any | None = None,
    subscription_class: Any | None = None
) -> DynamicModule

Create a GraphQLModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	Any | None	GraphQLConfig or ``None`` for framework defaults.
`query_class`	Any | None	Optional root Strawberry ``Query`` type.
`mutation_class`	Any | None	Optional root Strawberry ``Mutation`` type.
`subscription_class`	Any | None	Optional root Strawberry ``Subscription`` type.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

def stub(cls) -> DynamicModule

Return a no-op GraphQLModule suitable for unit testing.

Registers a GraphQLProvider with no schema types configured. Useful for testing application structure without a real GraphQL executor.

Returns

Type	Description
DynamicModule	A DynamicModule with noop GraphQL configuration.

---

GraphQLProvider

Provider for GraphQL functionality.

This provider integrates GraphQL capabilities into the Lexigram Framework, including schema building, execution, federation, and monitoring.

Features:

• GraphQL schema management and execution
• Apollo Federation support (subgraph)
• DataLoaderProtocol for batching and caching
• Query validation and depth limiting
• Metrics collection and tracing
• Subscription support
• Security features (rate limiting, auth)

Configuration: The provider is configured via GraphQLConfig in the application configuration. If no config is provided, sensible defaults are used.

Usage

app = Application()
app.add_provider(GraphQLProvider())

def __init__(
    config: GraphQLConfig | None = None,
    query_class: Any | None = None,
    mutation_class: Any | None = None,
    subscription_class: Any | None = None,
    priority: ProviderPriority = ProviderPriority.PRESENTATION
) -> None

Initialize the GraphQL provider.

Parameters

Parameter	Type	Description
`config`	GraphQLConfig | None	GraphQL configuration. If None, uses defaults.
`query_class`	Any | None	Optional GraphQL Query class.
`mutation_class`	Any | None	Optional GraphQL Mutation class.
`subscription_class`	Any | None	Optional GraphQL Subscription class.
`priority`	ProviderPriority	Provider priority for initialization order.

def from_config(
    cls,
    config: GraphQLConfig,
    **context: Any
) -> GraphQLProvider

Create a GraphQLProvider from config.

Context kwargs may include query_class, mutation_class, subscription_class.

def auto_discover(
    cls,
    *packages: str,
    config: GraphQLConfig | None = None,
    **kwargs: Any
) -> GraphQLProvider

Create a GraphQLProvider by scanning packages for Strawberry types.

Scans each package recursively for classes decorated with @strawberry.type whose name is Query, Mutation, or Subscription. Use @strawberry.type plus the naming convention or explicitly set __graphql_role__ = "query" / "mutation" / "subscription" on the class to override the name-based detection.

Parameters

Parameter	Type	Description
`config`	GraphQLConfig | None	Optional GraphQLConfig. Falls back to framework defaults when not provided. **kwargs: Extra keyword arguments forwarded to GraphQLProvider.__init__.

Returns

Type	Description
GraphQLProvider	A configured GraphQLProvider instance.

Example

app.add_provider(GraphQLProvider.auto_discover("my_app.graphql"))

In my_app/graphql/schema.py

import strawberry

@strawberry.type
class Query:
    @strawberry.field
    async def hello(self) -> str:
        return "world"

@strawberry.type
class Mutation:
    @strawberry.mutation
    async def set_name(self, name: str) -> str:
        return name

async def register(container: ContainerRegistrarProtocol) -> None

Register GraphQL services with the DI container.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	The dependency injection container registrar.

async def boot(container: BootContainerProtocol) -> None

Initialize GraphQL components.

Parameters

Parameter	Type	Description
`container`	BootContainerProtocol	The application container resolver.

def executor() -> GraphQLExecutorProtocol | None

Get the GraphQL executor instance.

property context_factory() -> ContextFactory | None

Get the GraphQL context factory instance.

async def shutdown() -> None

Shutdown GraphQL components.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check the health of GraphQL components.

Returns

Type	Description
HealthCheckResult	Health check results.

---

GraphQLRequest

GraphQL request model.

Represents an incoming GraphQL request with query, variables, and operation name.

Attributes: query: The GraphQL query string. variables: Variables for the query. operation_name: Name of the operation to execute. extensions: Optional extensions data.

---

GraphQLRequestReceivedHook

Payload fired when a GraphQL operation request is received.

Attributes: operation_type: "query", "mutation", or "subscription". operation_name: Named operation, if present in the document.

---

GraphQLResponse

GraphQL response model.

Represents a GraphQL response with data and/or errors.

Attributes: data: The result data. errors: List of errors if any occurred. extensions: Optional response extensions.

property has_errors() -> bool

Check if response has errors.

property is_successful() -> bool

Check if response is successful.

def add_error(
    message: str,
    path: list[str | int] | None = None,
    extensions: dict[str, Any] | None = None
) -> None

Add an error to the response.

Parameters

Parameter	Type	Description
`message`	str	Error message.
`path`	list[str | int] | None	Path to the error.
`extensions`	dict[str, Any] | None	Additional error data.

---

GraphQLResponsePreparedHook

Payload fired after a GraphQL response has been assembled.

Attributes: operation_type: "query", "mutation", or "subscription". has_errors: True if the response contains any GraphQL errors.

---

GraphQLSchemaBuiltHook

Payload fired after the GraphQL schema has been compiled and is ready.

---

GraphQLSubscriptionController

GraphQL WebSocket subscription controller.

This controller handles GraphQL subscriptions via WebSocket at the configured path (default: /graphql/subscriptions).

def __init__(provider: GraphQLProvider | None = None) -> None

Initialize the subscription controller.

Parameters

Parameter	Type	Description
`provider`	GraphQLProvider | None	GraphQL provider instance. If None, will be resolved from the DI container at request time.

def collect_routes(cls) -> list[dict[str, Any]]

Collect routes from controller methods.

---

InMemoryCache

In-memory cache implementation.

Simple in-memory cache with optional TTL support. Suitable for single-request caching in DataLoaders.

Example

cache = InMemoryCache[str, User](ttl_seconds=60)

cache.set("user:1", user)
user = cache.get("user:1")

def __init__(
    ttl_seconds: float = 0,
    max_size: int = 0
) -> None

Initialize the cache.

Parameters

Parameter	Type	Description
`ttl_seconds`	float	Time-to-live in seconds (0 for no TTL).
`max_size`	int	Maximum cache size (0 for unlimited).

property size() -> int

Get current cache size.

def get(key: K) -> V | None

Get a cached value.

Parameters

Parameter	Type	Description
`key`	K	Cache key.

Returns

Type	Description
V | None	Cached value or None if not found or expired.

def set(
    key: K,
    value: V
) -> None

Set a cached value.

Parameters

Parameter	Type	Description
`key`	K	Cache key.
`value`	V	Value to cache.

def has(key: K) -> bool

Check if key exists and is not expired.

Parameters

Parameter	Type	Description
`key`	K	Cache key.

Returns

Type	Description
bool	True if key exists and is valid.

def delete(key: K) -> None

Delete a cached value.

Parameters

Parameter	Type	Description
`key`	K	Cache key.

def clear() -> None

Clear all cached values.

def cleanup_expired() -> int

Remove expired entries.

Returns

Type	Description
int	Number of entries removed.

---

InMemoryPersistedQueryStore

In-memory implementation of PersistedQueryStore.

Warning: Not suitable for production use with multiple instances. Use RedisPersistedQueryStore for production deployments.

def __init__(ttl_seconds: int | None = None)

Initialize the store.

Parameters

Parameter	Type	Description
`ttl_seconds`	int | None	Optional time-to-live for entries in seconds.

async def get(hash: str) -> str | None

Get a query by its hash.

async def put(
    hash: str,
    query: str
) -> None

Store a query with its hash.

def clear() -> None

Clear all stored queries.

---

IntrospectionConfig

GraphQL introspection configuration.

Attributes: enabled: Whether introspection is enabled. Defaults to True but the executor automatically disables it in environments not listed in allowed_environments. allowed_environments: Set of environment names (matched against LEX_ENV or the app environment) where introspection is permitted. Defaults to {"development", "testing"}; production is intentionally excluded.

---

IntrospectionHandler

Handle GraphQL schema introspection.

Provides utilities for introspecting GraphQL schemas, extracting type information, and generating schema documentation.

Example

handler = IntrospectionHandler(schema)

# Get all types
types = await handler.get_types()

# Get fields for a type
fields = await handler.get_type_fields("User")

def __init__(
    schema: Schema,
    enabled: bool = True
) -> None

Initialize the handler.

Parameters

Parameter	Type	Description
`schema`	Schema	Strawberry GraphQL schema.
`enabled`	bool	Whether introspection is enabled.

property enabled() -> bool

Check if introspection is enabled.

def enable() -> None

Enable introspection.

def disable() -> None

Disable introspection.

async def introspect(simplified: bool = False) -> dict[str, Any]

Run introspection query.

Parameters

Parameter	Type	Description
`simplified`	bool	Use simplified query.

Returns

Type	Description
dict[str, Any]	Introspection result data.

async def get_types() -> list[dict[str, Any]]

Get all types in the schema.

Returns

Type	Description
list[dict[str, Any]]	List of type definitions.

async def get_type_fields(type_name: str) -> list[dict[str, Any]]

Get fields for a type.

Parameters

Parameter	Type	Description
`type_name`	str	Name of the type.

Returns

Type	Description
list[dict[str, Any]]	List of field definitions.

async def get_query_type() -> dict[str, Any] | None

Get the query type.

Returns

Type	Description
dict[str, Any] | None	Query type definition or None.

async def get_mutation_type() -> dict[str, Any] | None

Get the mutation type.

Returns

Type	Description
dict[str, Any] | None	Mutation type definition or None.

async def get_subscription_type() -> dict[str, Any] | None

Get the subscription type.

Returns

Type	Description
dict[str, Any] | None	Subscription type definition or None.

def clear_cache() -> None

Clear the introspection cache.

async def generate_sdl() -> str

Generate SDL (Schema Definition Language) from schema.

Returns

Type	Description
str	Schema as SDL string.

---

IsAdmin

Permission that requires admin role.

async def has_permission(
    source: Any,
    info: Info[Any, Any],
    **kwargs: Any
) -> bool

Check if user has admin role.

---

IsAuthenticated

Permission that requires user authentication.

async def has_permission(
    source: Any,
    info: Info[Any, Any],
    **kwargs: Any
) -> bool

Check if user is authenticated.

---

IsOwner

Permission that requires ownership of the resource.

async def has_permission(
    source: Any,
    info: Info[Any, Any],
    **kwargs: Any
) -> bool

Check if user owns the resource.

---

IsOwnerOrAdmin

Permission that requires ownership or admin role.

async def has_permission(
    source: Any,
    info: Info[Any, Any],
    **kwargs: Any
) -> bool

Check if user owns the resource or is admin.

---

LoaderCache

Abstract base class for DataLoaderProtocol caches.

Implement this interface to provide custom caching behavior for DataLoaders.

def get(key: K) -> V | None

Get a cached value.

Parameters

Parameter	Type	Description
`key`	K	Cache key.

Returns

Type	Description
V | None	Cached value or None.

def set(
    key: K,
    value: V
) -> None

Set a cached value.

Parameters

Parameter	Type	Description
`key`	K	Cache key.
`value`	V	Value to cache.

def has(key: K) -> bool

Check if key exists in cache.

Parameters

Parameter	Type	Description
`key`	K	Cache key.

Returns

Type	Description
bool	True if key exists.

def delete(key: K) -> None

Delete a cached value.

Parameters

Parameter	Type	Description
`key`	K	Cache key.

def clear() -> None

Clear all cached values.

---

MetricsCollectorProtocol

Collector for GraphQL metrics.

Collects and aggregates metrics from GraphQL operations. When a MetricsRecorderProtocol is provided (resolved from the DI container) every recorded query stat is forwarded to the kernel-level unified metrics pipeline so GraphQL telemetry lands alongside infra metrics.

Example

from lexigram.contracts.observability.metrics import MetricsRecorderProtocol

collector = MetricsCollectorProtocol(recorder=recorder)
collector.record(QueryStats(operation_name="GetUser", duration_ms=50.0))

def __init__(
    max_history: int = 1000,
    recorder: MetricsRecorderProtocol | None = None
) -> None

Initialize the collector.

Parameters

Parameter	Type	Description
`max_history`	int	Maximum number of stats to keep in the rolling history.
`recorder`	MetricsRecorderProtocol | None	Optional kernel MetricsRecorderProtocol for unified observability.

def record(stats: QueryStats) -> None

Record query statistics and forward to the kernel MetricsRecorderProtocol.

Parameters

Parameter	Type	Description
`stats`	QueryStats	Query statistics.

def record_error(error: Exception) -> None

Record an error.

Parameters

Parameter	Type	Description
`error`	Exception	The exception.

def get_metrics() -> GraphQLMetrics

Get current metrics.

Returns

Type	Description
GraphQLMetrics	Current metrics.

def get_recent_stats(limit: int = 100) -> list[QueryStats]

Get recent query statistics.

Parameters

Parameter	Type	Description
`limit`	int	Maximum number of stats to return.

Returns

Type	Description
list[QueryStats]	List of recent stats.

def reset() -> None

Reset all metrics.

async def close() -> None

Async close hook for the collector (no-op).

Provides symmetry with other resources that expose async close so callers can await shutdown without conditional checks.

---

MetricsConfig

Metrics collection configuration.

Attributes: enabled: Whether metrics are enabled. namespace: Metrics namespace/prefix. include_labels: Labels to include in metrics. histogram_buckets: Buckets for duration histograms.

---

MetricsExtension

Strawberry extension for metrics collection.

Automatically collects metrics for all GraphQL operations.

Example

from lexigram.graphql.monitoring import MetricsExtension

schema = strawberry.Schema(
    query=Query,
    extensions=[MetricsExtension()],
)

def __init__(collector: MetricsCollectorProtocol | None = None) -> None

Initialize the extension.

Parameters

Parameter	Type	Description
`collector`	MetricsCollectorProtocol | None	Metrics collector to use.

async def on_operation() -> AsyncGenerator[None, None]

Hook called during operation execution.

---

MutationResult

Standard mutation result type.

Attributes: success: Whether the mutation succeeded. data: Result data if successful. errors: Error messages if failed.

---

NoOpCache

No-operation cache (disables caching).

Use this when you want to disable DataLoaderProtocol caching while keeping the batching behavior.

def get(key: K) -> V | None

Always returns None.

def set(
    key: K,
    value: V
) -> None

Does nothing.

def has(key: K) -> bool

Always returns False.

def delete(key: K) -> None

Does nothing.

def clear() -> None

Does nothing.

---

OnErrorEvent

Emitted when a GraphQL operation encounters an unhandled exception.

This is fired for infrastructure-level errors (timeouts, executor crashes), not for user-facing field errors which are part of the normal GraphQL response.

---

OperationInfo

Information about a GraphQL operation.

---

OperationType

GraphQL operation types.

---

PageInfo

Pagination information.

---

PagedResult

Simple offset-based pagination result.

Attributes: items: List of items in the current page. total: Total number of items. page: Current page number (1-indexed). page_size: Number of items per page. has_next: Whether there are more pages. has_previous: Whether there are previous pages.

property total_pages() -> int

Calculate total number of pages.

---

PaginationInput

Input for pagination parameters.

Attributes: page: Page number (1-indexed). page_size: Number of items per page.

---

PlaygroundConfig

GraphQL Playground/GraphiQL configuration.

Attributes: enabled: Whether playground is enabled. path: Path to serve playground at. title: Title for the playground page.

---

QueryMetrics

Metrics for a GraphQL query execution.

def complete() -> None

Mark the query as complete and calculate duration.

---

QueryStats

Statistics for a single query execution.

Attributes: operation_name: Name of the operation. operation_type: Type (query, mutation, subscription). start_time: Execution start time. end_time: Execution end time. duration_ms: Execution duration in milliseconds. success: Whether execution succeeded. error_count: Number of errors.

---

RateLimitConfig

Configuration for rate limiting.

Attributes: enabled: Whether rate limiting is enabled. requests_per_minute: Number of allowed requests per minute. burst_limit: Maximum burst size.

---

RateLimitExtension

Enforce rate limiting on every GraphQL operation.

This extension delegates the actual limit check to the injected RateLimiter (resolved from the DI container by GraphQLProvider) and raises RateLimitError when the limit is exceeded.

Register it on the schema builder instead of wiring it into the executor

builder.add_extension(
    RateLimitExtension(rate_limiter=rate_limiter, max_requests=60)
)

Parameters

Parameter	Type	Description
`rate_limiter`		Rate limiter implementation; when ``None`` the extension is a no-op (useful for local/test environments).
`max_requests`		Request allowance per window.
`window_seconds`		Window length in seconds.

def __init__(
    *,
    rate_limiter: RateLimiter | None = None,
    max_requests: int = 60,
    window_seconds: int = 60
) -> None

async def on_operation() -> AsyncGenerator[None, None]

Check the rate limit before the GraphQL operation executes.

---

RateLimiter

Base class for rate limiters.

async def is_allowed(
    context: Any,
    **kwargs: Any
) -> bool

Check if a request is allowed.

---

RedisPersistedQueryStore

Redis-backed implementation of PersistedQueryStore.

Suitable for production use with multiple application instances.

def __init__(
    redis_client: Any,
    key_prefix: str = 'graphql:apq:',
    ttl_seconds: int = 86400
)

Initialize the store.

Parameters

Parameter	Type	Description
`redis_client`	Any	Redis client instance (aioRedis or redis-py async).
`key_prefix`	str	Prefix for Redis keys.
`ttl_seconds`	int	Time-to-live for entries in seconds.

async def get(hash: str) -> str | None

Get a query by its hash.

async def put(
    hash: str,
    query: str
) -> None

Store a query with its hash.

---

ResolverAdapter

Adapts a plain callable to the ResolverProtocol protocol.

Accepts any sync or async callable with the signature (parent, args, context, info) -> Any and exposes it as a ResolverProtocol via a resolve method. Sync callables are called directly; their return value is awaited only when it happens to be a coroutine.

Parameters

Parameter	Type	Description
`func`		A sync or async callable that implements the resolver logic. The callable should accept up to four positional arguments: ``(parent, args, context, info)``. Arguments beyond what the callable declares are silently dropped so that short signatures (e.g. ``lambda parent, args: parent.id``) work transparently.

def __init__(func: Any) -> None

async def resolve(
    parent: Any,
    args: dict[str, Any],
    context: Any,
    info: Any
) -> Any

Invoke the wrapped callable with the resolver arguments.

Parameters

Parameter	Type	Description
`parent`	Any	Parent (root) object for the field.
`args`	dict[str, Any]	Parsed field arguments from the GraphQL query.
`context`	Any	Shared execution context.
`info`	Any	GraphQL resolution info object.

Returns

Type	Description
Any	Resolved field value.

---

ResolverInfo

Context information passed to resolvers.

---

SchemaBuilderProtocol

Build GraphQL schemas with configuration.

Provides a fluent interface for constructing GraphQL schemas with proper configuration, extensions, and type registration.

Example

builder = SchemaBuilderProtocol()

schema = (
    builder
    .query(Query)
    .mutation(Mutation)
    .subscription(Subscription)
    .add_extension(QueryLogger())
    .build()
)

def __init__(config: GraphQLConfig | None = None) -> None

Initialize the schema builder.

Parameters

Parameter	Type	Description
`config`	GraphQLConfig | None	GraphQL configuration.

def query(query_type: type[Any]) -> SchemaBuilderProtocol

Set the query type.

Parameters

Parameter	Type	Description
`query_type`	type[Any]	The query type class.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def mutation(mutation_type: type[Any]) -> SchemaBuilderProtocol

Set the mutation type.

Parameters

Parameter	Type	Description
`mutation_type`	type[Any]	The mutation type class.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def subscription(subscription_type: type[Any]) -> SchemaBuilderProtocol

Set the subscription type.

Parameters

Parameter	Type	Description
`subscription_type`	type[Any]	The subscription type class.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def add_type(type_class: type[Any]) -> SchemaBuilderProtocol

Add an additional type to the schema.

Parameters

Parameter	Type	Description
`type_class`	type[Any]	The type class to add.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def add_types(*type_classes: type[Any]) -> SchemaBuilderProtocol

Add multiple types to the schema.

Parameters

Parameter	Type	Description
`type_classes`	type[Any]	Type classes to add.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def add_extension(extension: Any) -> SchemaBuilderProtocol

Add a schema extension.

Parameters

Parameter	Type	Description
`extension`	Any	The extension to add (kept as Any to avoid strict coupling to Strawberry's extension type in various environments).

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def add_dataloader(
    name: str,
    factory: Any
) -> SchemaBuilderProtocol

Register a DataLoaderProtocol factory for per-request loader initialisation.

Stored factories are wired into ContextFactory by GraphQLProvider during boot(), so loaders are available inside resolvers via context.get_dataloader(name).

Parameters

Parameter	Type	Description
`name`	str	Unique loader name.
`factory`	Any	Callable ``(context) -> DataLoaderProtocol`` invoked per-request.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def add_directive(directive: Any) -> SchemaBuilderProtocol

Add a custom directive.

Parameters

Parameter	Type	Description
`directive`	Any	The directive to add.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def scalar_override(
    original: Any,
    override: Any
) -> SchemaBuilderProtocol

Override a scalar type.

Parameters

Parameter	Type	Description
`original`	Any	Original scalar type.
`override`	Any	Override scalar type.

Returns

Type	Description
SchemaBuilderProtocol	Self for chaining.

def build() -> Schema

Build the GraphQL schema.

Returns

Type	Description
Schema	Configured Strawberry schema.

Raises

Exception	Description
ValueError	If no query type is set and no default is provided.

property config() -> GraphQLConfig

Get the configuration.

---

SchemaBuiltEvent

Emitted after the GraphQL schema is successfully built.

Consumers may inspect type counts for observability and diagnostics.

---

SchemaValidator

Validate GraphQL schemas and queries.

Provides comprehensive validation of GraphQL queries against the schema, delegating depth enforcement to DepthLimitValidator and alias enforcement to AliasLimitValidator.

Example

validator = SchemaValidator(schema)
result = validator.validate_query(query)

if not result.is_valid:
    for error in result.errors:
        logger.info("Validation error: %s", error)

def __init__(
    schema: Any,
    max_depth: int = 10,
    max_aliases: int = 10
) -> None

Initialize the validator.

Parameters

Parameter	Type	Description
`schema`	Any	GraphQL schema.
`max_depth`	int	Maximum query depth (delegated to DepthLimitValidator).
`max_aliases`	int	Maximum number of aliases (delegated to AliasLimitValidator).

def validate_query(
    query: str,
    variables: dict[str, Any] | None = None
) -> ValidationResult

Validate a GraphQL query.

Parameters

Parameter	Type	Description
`query`	str	GraphQL query string.
`variables`	dict[str, Any] | None	Query variables.

Returns

Type	Description
ValidationResult	Validation result.

---

SortInput

Input for sorting.

Attributes: field: Field to sort by. direction: Sort direction (ASC or DESC).

---

SubscriptionConfig

WebSocket subscription configuration.

Attributes: enabled: Whether subscriptions are enabled. path: WebSocket path for subscriptions. protocol: WebSocket protocol to use. keepalive_interval: Keepalive interval in seconds. connection_timeout: Connection timeout in seconds.

---

SubscriptionInfo

Information about an active subscription.

---

SubscriptionProtocol

WebSocket subscription protocols.

---

SubscriptionStartedEvent

Emitted when a GraphQL subscription is established.

Provides the subscription identifier and operation name for tracking and auditing long-lived connections.

---

TraceSpan

A span in the execution trace.

Attributes: name: Span name. start_time: Start timestamp. end_time: End timestamp. duration_ms: Duration in milliseconds. metadata: Additional span metadata. children: Child spans.

def end() -> None

End the span and calculate duration.

def add_child(child: TraceSpan) -> None

Add a child span.

def to_dict() -> dict[str, Any]

Convert to dictionary.

Returns

Type	Description
dict[str, Any]	Dictionary representation.

---

TracingConfig

Distributed tracing configuration.

Attributes: enabled: Whether tracing is enabled. service_name: Service name for tracing. trace_resolvers: Whether to trace individual resolvers. trace_dataloaders: Whether to trace DataLoaderProtocol batches. sample_rate: Sampling rate (0.0 to 1.0).

---

TracingExtension

Strawberry extension for query tracing.

Adds tracing information to query responses in Apollo tracing format.

Example

from lexigram.graphql.monitoring import TracingExtension

schema = strawberry.Schema(
    query=Query,
    extensions=[TracingExtension()],
)

def __init__(
    include_in_response: bool = True,
    tracer: TracerProtocol | None = None
) -> None

Initialize the extension.

Parameters

Parameter	Type	Description
`include_in_response`	bool	Include tracing in response extensions.
`tracer`	TracerProtocol | None	Optional kernel TracerProtocol for unified distributed tracing. When provided, GraphQL operation and resolver spans are forwarded to the application-wide tracing pipeline so GraphQL telemetry is correlated with infra traces.

def on_operation() -> Iterator[None]

Hook called during operation execution.

def resolve(
    _next: Callable[Ellipsis, Any],
    root: Any,
    info: Any,
    *args: Any,
    **kwargs: Any
) -> Any

Hook called during field resolution.

---

UnifiedRateLimiter

Rate limiter that delegates to an injected web rate limiter.

def __init__(web_rate_limiter: WebRateLimiterProtocol | None = None) -> None

async def is_allowed(
    context: Any,
    **kwargs: Any
) -> bool

Check if a request is allowed using the web rate limiter.

---

ValidationResult

Result of query validation.

Attributes: is_valid: Whether the query is valid. errors: List of validation errors. depth: Maximum query depth. field_count: Total number of fields. warnings: Non-fatal warnings.

def add_error(error: str) -> None

Add a validation error.

def add_warning(warning: str) -> None

Add a validation warning.

---

Functions

compute_query_hash

def compute_query_hash(query: str) -> str

Compute SHA256 hash of a GraphQL query.

Parameters

Parameter	Type	Description
`query`	str	The GraphQL query string.

Returns

Type	Description
str	Hex-encoded SHA256 hash.

---

create_connection_type

def create_connection_type(
    node_type: type[T],
    name: str | None = None
) -> type[Connection[T]]

Create a Connection type for a specific node type.

Parameters

Parameter	Type	Description
`node_type`	type[T]	The node type for the connection.
`name`	str | None	Optional name prefix for the connection type.

Returns

Type	Description
type[Connection[T]]	A Connection type class.

---

create_depth_limit

def create_depth_limit(
    max_depth: int = 10,
    ignore_introspection: bool = True
) -> DepthLimitExtension

Create a depth limit extension (convenience function).

Parameters

Parameter	Type	Description
`max_depth`	int	Maximum allowed depth.
`ignore_introspection`	bool	Skip introspection queries.

Returns

Type	Description
DepthLimitExtension	Configured extension.

Example

schema = strawberry.Schema(
    query=Query,
    extensions=[create_depth_limit(5)],
)

---

create_loader

def create_loader(
    batch_fn: Callable[[list[K]], Awaitable[list[V | None]]],
    batch_size: int = 100,
    cache_enabled: bool = True
) -> DataLoaderProtocol[K, V]

Create a DataLoaderProtocol (convenience function).

Parameters

Parameter	Type	Description
`batch_fn`	Callable[[list[K]], Awaitable[list[V | None]]]	Batch load function.
`batch_size`	int	Maximum batch size.
`cache_enabled`	bool	Whether to enable caching.

Returns

Type	Description
DataLoaderProtocol[K, V]	Configured DataLoaderProtocol.

Example

user_loader = create_loader(
    batch_load_users,
    batch_size=50,
)

user = await user_loader.load("123")

---

create_schema

def create_schema(
    query: type[Any],
    mutation: type[Any] | None = None,
    subscription: type[Any] | None = None,
    types: list[type[Any]] | None = None,
    extensions: list[Any] | None = None,
    config: GraphQLConfig | None = None
) -> Schema

Create a GraphQL schema (convenience function).

Parameters

Parameter	Type	Description
`query`	type[Any]	Query type class.
`mutation`	type[Any] | None	Optional mutation type class.
`subscription`	type[Any] | None	Optional subscription type class.
`types`	list[type[Any]] | None	Additional type classes.
`extensions`	list[Any] | None	Schema extensions.
`config`	GraphQLConfig | None	GraphQL configuration.

Returns

Type	Description
Schema	Configured Strawberry schema.

Example

schema = create_schema(
    query=Query,
    mutation=Mutation,
    config=GraphQLConfig(depth_limit=5),
)

---

decode_cursor

def decode_cursor(cursor: str) -> int

Decode a base64 cursor to an offset.

Parameters

Parameter	Type	Description
`cursor`	str	The cursor string to decode.

Returns

Type	Description
int	The decoded offset.

Raises

Exception	Description
ValueError	If the cursor is invalid.

---

decode_cursor_to_id

def decode_cursor_to_id(cursor: str) -> str

Decode a cursor to an item ID.

Parameters

Parameter	Type	Description
`cursor`	str	The cursor string to decode.

Returns

Type	Description
str	The decoded item ID.

Raises

Exception	Description
ValueError	If the cursor is invalid.

---

encode_cursor

def encode_cursor(offset: int) -> str

Encode an offset as a base64 cursor.

Parameters

Parameter	Type	Description
`offset`	int	The offset to encode.

Returns

Type	Description
str	Base64-encoded cursor string.

---

encode_cursor_from_id

def encode_cursor_from_id(item_id: str) -> str

Encode an item ID as an opaque cursor.

Parameters

Parameter	Type	Description
`item_id`	str	The item ID to encode.

Returns

Type	Description
str	Base64-encoded cursor string.

---

execute_query

async def execute_query(
    schema: StrawberrySchema,
    query: str,
    variables: dict[str, Any] | None = None,
    operation_name: str | None = None,
    context: GraphQLContext | None = None
) -> GraphQLResponse[Any]

Execute a GraphQL query (convenience function).

If you need to build a context from a plain dictionary, use ContextFactory.from_dict() before calling this function.

Parameters

Parameter	Type	Description
`schema`	StrawberrySchema	Strawberry GraphQL schema.
`query`	str	GraphQL query string.
`variables`	dict[str, Any] | None	Query variables.
`operation_name`	str | None	Operation name.
`context`	GraphQLContext | None	GraphQL context.

Returns

Type	Description
GraphQLResponse[Any]	GraphQL response.

---

field

def field(
    name: str | None = None,
    description: str | None = None,
    deprecation_reason: str | None = None,
    default: Any = strawberry.UNSET,
    default_factory: Callable[[], Any] | None = None,
    permission_classes: list[Any] | None = None
) -> Any

Define a GraphQL field.

This is a wrapper around Strawberry’s field function with additional Lexigram-specific functionality.

Parameters

Parameter	Type	Description
`name`	str | None	Optional field name.
`description`	str | None	Field description.
`deprecation_reason`	str | None	Deprecation reason if deprecated.
`default`	Any	Default value.
`default_factory`	Callable[[], Any] | None	Factory for default value.
`permission_classes`	list[Any] | None	Permission classes for authorization.

Returns

Type	Description
Any	Strawberry field.

Example

@strawberry.type
class User:
    id: str
    name: str = field(description="User's full name")
    email: str = field(description="User's email address")

---

get_context

def get_context(info: Info) -> GraphQLContext

Get the GraphQL context from resolver info.

Parameters

Parameter	Type	Description
`info`	Info	Strawberry resolver info.

Returns

Type	Description
GraphQLContext	GraphQL context.

Example

@strawberry.type
class Query:
    @strawberry.field
    async def me(self, info: Info) -> User:
        context = get_context(info)
        return context.user

---

get_introspection_query

def get_introspection_query(simplified: bool = False) -> str

Get the GraphQL introspection query.

Parameters

Parameter	Type	Description
`simplified`	bool	If True, return simplified query.

Returns

Type	Description
str	Introspection query string.

---

get_metrics_collector

async def get_metrics_collector(context: Any | None = None) -> MetricsCollectorProtocol

Get the metrics collector instance.

---

log_resolver

def log_resolver(fn: F) -> F

Wrap an async GraphQL resolver with structured entry/exit/error logging.

Emits debug-level log events on entry and exit, and an error-level event when an exception propagates out of the resolver. The exception is always re-raised — this decorator never swallows errors.

Parameters

Parameter	Type	Description
`fn`	F	The async resolver function to wrap.

Returns

Type	Description
F	Wrapped resolver with structured logging applied.

Example

@strawberry.type
class Mutation:
    @log_resolver
    async def create_user(self, info: Info, input: CreateUserInput) -> User:
        return await service.create(input)

---

mutation

def mutation(
    name: str | None = None,
    description: str | None = None,
    deprecation_reason: str | None = None,
    permission_classes: list[Any] | None = None
) -> Callable[[Callable[P, T]], Callable[P, Any]]

Decorator for marking a method as a GraphQL mutation.

This is a wrapper around Strawberry’s mutation decorator with additional Lexigram-specific functionality.

Parameters

Parameter	Type	Description
`name`	str | None	Optional field name (defaults to function name).
`description`	str | None	Field description for documentation.
`deprecation_reason`	str | None	If set, marks the field as deprecated.
`permission_classes`	list[Any] | None	List of permission classes for authorization.

Returns

Type	Description
Callable[[Callable[P, T]], Callable[P, Any]]	Decorated function.

Example

@strawberry.type
class Mutation:
    @mutation(description="Create a new user")
    async def create_user(
        self, info: Info, input: CreateUserInput
    ) -> User:
        return await create_user(input)

---

query

def query(
    name: str | None = None,
    description: str | None = None,
    deprecation_reason: str | None = None,
    permission_classes: list[Any] | None = None
) -> Callable[[Callable[P, T]], Callable[P, Any]]

Decorator for marking a method as a GraphQL query.

This is a wrapper around Strawberry’s field decorator with additional Lexigram-specific functionality.

Parameters

Parameter	Type	Description
`name`	str | None	Optional field name (defaults to function name).
`description`	str | None	Field description for documentation.
`deprecation_reason`	str | None	If set, marks the field as deprecated.
`permission_classes`	list[Any] | None	List of permission classes for authorization.

Returns

Type	Description
Callable[[Callable[P, T]], Callable[P, Any]]	Decorated function.

Example

@strawberry.type
class Query:
    @query(description="Get user by ID")
    async def user(self, info: Info, id: str) -> User:
        return await get_user(id)

---

resolver

def resolver(
    name: str | None = None,
    description: str | None = None
) -> Callable[[Callable[P, T]], Callable[P, Any]]

Decorator for field resolvers.

Use this to define a resolver function that can be used as a field resolver in a GraphQL type.

Parameters

Parameter	Type	Description
`name`	str | None	Optional resolver name.
`description`	str | None	ResolverProtocol description.

Returns

Type	Description
Callable[[Callable[P, T]], Callable[P, Any]]	Decorated resolver function.

Example

@resolver(description="Resolve user's full name")
async def resolve_full_name(
    root: User, info: Info
) -> str:
    return f"{root.first_name} {root.last_name}"

@strawberry.type
class User:
    first_name: str
    last_name: str
    full_name: str = strawberry.field(resolver=resolve_full_name)

---

retry_resolver

def retry_resolver(
    max_retries: int = 3,
    *,
    delay: float = 0.1,
    exceptions: tuple[type[Exception], Ellipsis] = (Exception,)
) -> Callable[[F], F]

Wrap an async GraphQL resolver with automatic retry logic.

Retries the decorated resolver up to max_retries times when any of the specified exceptions are raised. Uses exponential back-off between attempts: delay * 2 ** attempt seconds.

Parameters

Parameter	Type	Description
`max_retries`	int	Maximum number of attempts before re-raising the last exception. Must be at least 1.
`delay`	float	Base delay in seconds between retries. Doubles on each attempt.
`exceptions`	tuple[type[Exception], Ellipsis]	Tuple of exception types that trigger a retry. Defaults to ``(Exception,)`` — any exception retries.

Returns

Type	Description
Callable[[F], F]	Decorator that wraps an async resolver with retry machinery.

Example

@strawberry.type
class Query:
    @retry_resolver(max_retries=3, delay=0.2, exceptions=(TimeoutError,))
    async def user(self, info: Info, id: str) -> User:
        return await fetch_user(id)

---

subscription

def subscription(
    name: str | None = None,
    description: str | None = None,
    deprecation_reason: str | None = None,
    permission_classes: list[Any] | None = None
) -> Callable[[Callable[P, T]], Callable[P, Any]]

Decorator for marking a method as a GraphQL subscription.

This is a wrapper around Strawberry’s subscription decorator with additional Lexigram-specific functionality.

Parameters

Parameter	Type	Description
`name`	str | None	Optional field name (defaults to function name).
`description`	str | None	Field description for documentation.
`deprecation_reason`	str | None	If set, marks the field as deprecated.
`permission_classes`	list[Any] | None	List of permission classes for authorization.

Returns

Type	Description
Callable[[Callable[P, T]], Callable[P, Any]]	Decorated function.

Example

@strawberry.type
class Subscription:
    @subscription(description="Subscribe to messages")
    async def messages(
        self, info: Info, room_id: str
    ) -> AsyncGenerator[Message, None]:
        async for msg in message_stream(room_id):
            yield msg

---

trace_resolver

def trace_resolver(name: str | None = None) -> Callable[[Callable[P, T]], Callable[P, T]]

Decorator for tracing resolver execution.

Parameters

Parameter	Type	Description
`name`	str | None	Optional span name.

Returns

Type	Description
Callable[[Callable[P, T]], Callable[P, T]]	Decorator function.

Example

@strawberry.type
class Query:
    @trace_resolver("fetch_user")
    @strawberry.field
    async def user(self, info: Info, id: str) -> User:
        return await get_user(id)

---

union_type

def union_type(
    *types: Any,
    name: str | None = None
) -> Any

Create a GraphQL union type.

Parameters

Parameter	Type	Description
`types`	Any	Types to include in the union.
`name`	str | None	Optional name for the union.

Returns

Type	Description
Any	Union type annotation.

Example

SearchResult = union_type(User, Post, Comment, name="SearchResult")

@strawberry.type
class Query:
    @strawberry.field
    def search(self, query: str) -> list[SearchResult]:
        ...

---

validate_query

def validate_query(
    schema: Any,
    query: str,
    variables: dict[str, Any] | None = None,
    max_depth: int = 10
) -> ValidationResult

Validate a GraphQL query (convenience function).

Parameters

Parameter	Type	Description
`schema`	Any	GraphQL schema.
`query`	str	GraphQL query string.
`variables`	dict[str, Any] | None	Query variables.
`max_depth`	int	Maximum allowed depth.

Returns

Type	Description
ValidationResult	Validation result.

---

Exceptions

AuthenticationError

Raised when authentication is required but missing.

Also satisfies lexigram.contracts.exceptions.AuthenticationError so middleware that catches the contracts type will catch this as well.

---

AuthorizationError

Raised when user is authenticated but lacks permissions.

Also satisfies lexigram.contracts.exceptions.AuthorizationError so middleware that catches the contracts type will catch this as well.

---

ForbiddenError

Raised when access is forbidden.

---

GraphQLConnectionError

Raised when a network connection error occurs during GraphQL execution.

---

GraphQLError

Base exception for all GraphQL errors.

def __init__(
    message: str,
    *,
    code: GraphQLErrorCode | None = None,
    **kwargs: Any
) -> None

---

InputGraphQLError

Raised for invalid user input.

---

NotFoundError

Raised when a requested resource is not found.

---

ParseError

Raised when query parsing fails.

---

QueryTooComplexError

Raised when query exceeds complexity limit.

---

QueryTooDeepError

Raised when a GraphQL query exceeds the maximum depth.

---

RateLimitError

Raised when rate limit is exceeded.

---

ResolverError

Raised when a resolver fails.

---

SubscriptionError

Raised for subscription-related errors.

---