Python API Reference

Python API Reference v0.15.6-rc.22¶

Functions¶

schema_query_only()¶

Create a simple schema configuration with only Query type.

This is a convenience function for schemas that only have queries.

Returns:

A QueryOnlyConfig with default settings

Signature:

def schema_query_only() -> QueryOnlyConfig

Returns: QueryOnlyConfig

schema_query_mutation()¶

Create a schema configuration with Query and Mutation types.

This is a convenience function for schemas with queries and mutations but no subscriptions.

Returns:

A QueryMutationConfig with default settings

Signature:

def schema_query_mutation() -> QueryMutationConfig

Returns: QueryMutationConfig

schema_full()¶

Create a schema configuration with all three root types.

This is a convenience function for fully-featured schemas.

Returns:

A FullSchemaConfig with default settings

Signature:

def schema_full() -> FullSchemaConfig

Returns: FullSchemaConfig

Types¶

ApiKeyConfig¶

API Key authentication configuration

App¶

Spikard application builder.

Methods¶

new()¶

Create a new application with the default server configuration.

Signature:

@staticmethod
def new() -> App

merge_axum_router()¶

Attach an existing Axum router to this application, returning ownership.

Signature:

def merge_axum_router(self, router: str) -> App

attach_axum_router()¶

Attach an Axum router using a mutable reference for incremental configuration.

Signature:

def attach_axum_router(self, router: str) -> App

into_router()¶

Build the underlying Axum router.

Errors:

Returns an error if server or router construction fails.

Signature:

def into_router(self) -> str

into_router_and_config()¶

Decompose the application into its Axum router and server configuration.

This is the low-level escape hatch used by the C FFI layer to start the server on a background thread while retaining the bind address from the caller-supplied ServerConfig. Prefer App.run for normal use.

Errors:

Returns an error if router construction fails.

Signature:

def into_router_and_config(self) -> str

default()¶

Signature:

@staticmethod
def default() -> App

AsyncApiConfig¶

AsyncAPI HTTP endpoint configuration

BackgroundJobMetadata¶

Methods¶

default()¶

Signature:

@staticmethod
def default() -> BackgroundJobMetadata

BackgroundTaskConfig¶

Configuration for in-process background task execution.

Methods¶

default()¶

Signature:

@staticmethod
def default() -> BackgroundTaskConfig

CompressionConfig¶

Compression configuration shared across runtimes

Methods¶

default()¶

Signature:

@staticmethod
def default() -> CompressionConfig

ContactInfo¶

Contact information

CorsConfig¶

CORS configuration for a route

Methods¶

allowed_methods_joined()¶

Get the cached joined methods string for preflight responses

Signature:

def allowed_methods_joined(self) -> str

allowed_headers_joined()¶

Get the cached joined headers string for preflight responses

Signature:

def allowed_headers_joined(self) -> str

is_origin_allowed()¶

Check if an origin is allowed (O(1) with wildcard, O(n) for exact match)

Signature:

def is_origin_allowed(self, origin: str) -> bool

is_method_allowed()¶

Check if a method is allowed (O(1) with wildcard, O(n) for exact match)

Signature:

def is_method_allowed(self, method: str) -> bool

default()¶

Signature:

@staticmethod
def default() -> CorsConfig

FullSchemaConfig¶

Configuration for fully-featured schemas with Query, Mutation, and Subscription types

Methods¶

default()¶

Signature:

@staticmethod
def default() -> FullSchemaConfig

GraphQlRouteConfig¶

Configuration for GraphQL routes

Provides a builder pattern for configuring GraphQL route parameters for the Spikard HTTP server's routing system.

Methods¶

new()¶

Create a new GraphQL route configuration with defaults

Default values:

• path: "/graphql"
• method: "POST"
• enable_playground: false

Signature:

@staticmethod
def new() -> GraphQlRouteConfig

path()¶

Set the HTTP path for the GraphQL endpoint

Signature:

def path(self, path: str) -> GraphQlRouteConfig

method()¶

Set the HTTP method for the GraphQL endpoint

Signature:

def method(self, method: str) -> GraphQlRouteConfig

enable_playground()¶

Enable or disable the GraphQL Playground UI

Signature:

def enable_playground(self, enable: bool) -> GraphQlRouteConfig

description()¶

Set a custom description for documentation

Signature:

def description(self, description: str) -> GraphQlRouteConfig

get_path()¶

Get the configured path

Signature:

def get_path(self) -> str

get_method()¶

Get the configured method

Signature:

def get_method(self) -> str

is_playground_enabled()¶

Check if playground is enabled

Signature:

def is_playground_enabled(self) -> bool

get_description()¶

Get the description if set

Signature:

def get_description(self) -> str | None

default()¶

Signature:

@staticmethod
def default() -> GraphQlRouteConfig

GraphQlSubscriptionSnapshot¶

Snapshot of a GraphQL subscription exchange over WebSocket.

GrpcConfig¶

Configuration for gRPC support

Controls how the server handles gRPC requests, including compression, timeouts, and protocol settings.

Stream Limits¶

This configuration enforces message-level size limits but delegates concurrent stream limiting to the HTTP/2 transport layer:

• Message Size Limits: The max_message_size field is enforced per individual message (request or response) in both unary and streaming RPCs. When a single message exceeds this limit, the request is rejected with PAYLOAD_TOO_LARGE (HTTP 413).

• Concurrent Stream Limits: The max_concurrent_streams is an advisory configuration passed to the HTTP/2 layer for connection-level stream negotiation. The HTTP/2 transport automatically enforces this limit and returns GOAWAY frames when exceeded. Applications should not rely on custom enforcement of this limit.

• Stream Response Size Limits: The max_stream_response_bytes field caps the total encoded bytes emitted across a server-streaming or bidi-streaming response. When the cumulative size exceeds the limit, the stream is terminated with tonic.Status.resource_exhausted. Defaults to None (unbounded).

Methods¶

default()¶

Signature:

@staticmethod
def default() -> GrpcConfig

Handler¶

Handler trait that all language bindings must implement

This trait is completely language-agnostic. Each binding (Python, Node, WASM) implements this trait to bridge their runtime to our HTTP server.

Methods¶

call()¶

Handle an HTTP request

Takes the extracted request data and returns a future that resolves to either:

• Ok(Response): A successful HTTP response
• Err((StatusCode, String)): An error with status code and message

Signature:

def call(self, request: Request, request_data: RequestData) -> HandlerResult

prefers_raw_json_body()¶

Whether this handler prefers consuming RequestData.raw_body over the parsed RequestData.body for JSON requests.

When True, the server may skip eager JSON parsing when there is no request-body schema validator attached to the route.

Signature:

def prefers_raw_json_body(self) -> bool

prefers_parameter_extraction()¶

Whether this handler wants to perform its own parameter validation/extraction (path/query/header/cookie).

When True, the server will skip ParameterValidator.validate_and_extract in ValidatingHandler. This is useful for language bindings which need to transform validated parameters into language-specific values (e.g., Python kwargs) without duplicating work. When False, the server stores validated output in RequestData.validated_params.

Signature:

def prefers_parameter_extraction(self) -> bool

wants_headers()¶

Whether this handler needs the parsed headers map in RequestData.

When False, the server may skip building RequestData.headers for requests without a body. (Requests with bodies still typically need Content-Type decisions.)

Signature:

def wants_headers(self) -> bool

wants_cookies()¶

Whether this handler needs the parsed cookies map in RequestData.

When False, the server may skip parsing cookies for requests without a body.

Signature:

def wants_cookies(self) -> bool

wants_request_extensions()¶

Whether this handler needs RequestData stored in request extensions.

When False, the server avoids inserting RequestData into extensions to skip cloning in hot paths.

Signature:

def wants_request_extensions(self) -> bool

static_response()¶

Return a pre-built static response if this handler always produces the same output. When Some, the server bypasses the full middleware pipeline and serves the pre-built response directly.

Signature:

def static_response(self) -> StaticResponse | None

IntoHandler¶

Convert user-facing handler functions into the low-level Handler trait.

Methods¶

into_handler()¶

Convert this value into a shared request handler.

Signature:

def into_handler(self) -> Handler

JsonRpcConfig¶

JSON-RPC server configuration

Methods¶

default()¶

Signature:

@staticmethod
def default() -> JsonRpcConfig

JsonRpcMethodInfo¶

JSON-RPC method metadata for routes that support JSON-RPC

This struct captures the metadata needed to expose HTTP routes as JSON-RPC methods, enabling discovery and documentation of RPC-compatible endpoints.

JwtConfig¶

JWT authentication configuration

LicenseInfo¶

License information

OpenApiConfig¶

OpenAPI configuration

Methods¶

default()¶

Signature:

@staticmethod
def default() -> OpenApiConfig

ParseRequest¶

Request body for POST /asyncapi/parse

ParseResult¶

Full parse result returned by POST /asyncapi/parse

ParsedChannel¶

A single channel extracted from an AsyncAPI spec

ParsedMessage¶

A resolved message (name + JSON Schema)

ParsedOperation¶

A single operation extracted from an AsyncAPI spec

ProblemDetails¶

RFC 9457 Problem Details for HTTP APIs

A machine-readable format for specifying errors in HTTP API responses. Per RFC 9457, all fields are optional. The type field defaults to "about:blank" if not specified.

Content-Type¶

Responses using this struct should set:

Content-Type: application/problem+json

{
  "type": "<https://spikard.dev/errors/validation-error",>
  "title": "Request Validation Failed",
  "status": 422,
  "detail": "2 validation errors in request body",
  "errors": [...]
}

Methods¶

with_detail()¶

Set the detail field

Signature:

def with_detail(self, detail: str) -> ProblemDetails

with_instance()¶

Set the instance field

Signature:

def with_instance(self, instance: str) -> ProblemDetails

not_found()¶

Create a not found error

Signature:

@staticmethod
def not_found(detail: str) -> ProblemDetails

method_not_allowed()¶

Create a method not allowed error

Signature:

@staticmethod
def method_not_allowed(detail: str) -> ProblemDetails

internal_server_error()¶

Create an internal server error

Signature:

@staticmethod
def internal_server_error(detail: str) -> ProblemDetails

bad_request()¶

Create a bad request error

Signature:

@staticmethod
def bad_request(detail: str) -> ProblemDetails

to_json()¶

Serialize to JSON string

Errors: Returns an error if the serialization fails.

Signature:

def to_json(self) -> str

to_json_pretty()¶

Serialize to pretty JSON string

Errors: Returns an error if the serialization fails.

Signature:

def to_json_pretty(self) -> str

QueryMutationConfig¶

Configuration for schemas with Query and Mutation types

Methods¶

default()¶

Signature:

@staticmethod
def default() -> QueryMutationConfig

QueryOnlyConfig¶

Configuration for schemas with only Query type

Methods¶

default()¶

Signature:

@staticmethod
def default() -> QueryOnlyConfig

RateLimitConfig¶

Rate limiting configuration shared across runtimes

Methods¶

default()¶

Signature:

@staticmethod
def default() -> RateLimitConfig

Request¶

Response¶

HTTP Response with custom status code, headers, and content

Methods¶

set_header()¶

Set a header

Signature:

def set_header(self, key: str, value: str) -> None

set_cookie()¶

Set a cookie in the response

Signature:

def set_cookie(self, key: str, value: str, secure: bool, http_only: bool, max_age: int, domain: str, path: str, same_site: str) -> None

default()¶

Signature:

@staticmethod
def default() -> Response

ResponseSnapshot¶

Snapshot of an Axum response used by higher-level language bindings.

Methods¶

text()¶

Return response body as UTF-8 string.

Signature:

def text(self) -> str

header()¶

Lookup header by case-insensitive name.

Signature:

def header(self, name: str) -> str | None

RouteBuilder¶

Builder for defining a route.

Methods¶

new()¶

Create a new builder for the provided HTTP method and path.

Signature:

@staticmethod
def new(method: Method, path: str) -> RouteBuilder

handler_name()¶

Assign an explicit handler name.

Signature:

def handler_name(self, name: str) -> RouteBuilder

request_schema_json()¶

Provide a raw JSON schema for the request body.

Signature:

def request_schema_json(self, schema: dict[str, Any]) -> RouteBuilder

response_schema_json()¶

Provide a raw JSON schema for the response body.

Signature:

def response_schema_json(self, schema: dict[str, Any]) -> RouteBuilder

params_schema_json()¶

Provide a raw JSON schema for request parameters.

Signature:

def params_schema_json(self, schema: dict[str, Any]) -> RouteBuilder

file_params_json()¶

Provide multipart file parameter configuration.

Signature:

def file_params_json(self, schema: dict[str, Any]) -> RouteBuilder

cors()¶

Attach a CORS configuration for this route.

Signature:

def cors(self, cors: CorsConfig) -> RouteBuilder

compression()¶

Attach a compression configuration for this route.

Signature:

def compression(self, compression: CompressionConfig) -> RouteBuilder

sync()¶

Mark the route as synchronous.

Signature:

def sync(self) -> RouteBuilder

handler_dependencies()¶

Declare the dependency keys that must be resolved before this handler runs.

Signature:

def handler_dependencies(self, dependencies: list[str]) -> RouteBuilder

SchemaConfig¶

Configuration for GraphQL schema building.

Encapsulates all schema-level configuration options including introspection control, complexity limits, and depth limits.

Methods¶

default()¶

Signature:

@staticmethod
def default() -> SchemaConfig

ServerConfig¶

Server configuration

Methods¶

default()¶

Signature:

@staticmethod
def default() -> ServerConfig

ServerInfo¶

Server information

SseEvent¶

An individual SSE event

Represents a single Server-Sent Event to be sent to a connected client. Events can have an optional type, ID, and retry timeout for advanced scenarios.

SSE Format¶

Events are serialized to the following text format:

event: event_type
data: {"json":"value"}
id: event-123
retry: 3000

Methods¶

with_id()¶

Set the event ID for client-side reconnection support

Sets an ID that clients can use to resume from this point if they disconnect. The client sends this ID back in the Last-Event-ID header when reconnecting.

Signature:

def with_id(self, id: str) -> SseEvent

with_retry()¶

Set the retry timeout for client reconnection

Sets the time in milliseconds clients should wait before attempting to reconnect if the connection is lost. The client browser will automatically handle reconnection.

Signature:

def with_retry(self, retry_ms: int) -> SseEvent

StaticFilesConfig¶

Static file serving configuration

TestClient¶

Core test client for making HTTP requests to a Spikard application.

This struct wraps axum-test's TestServer and provides a language-agnostic interface for making HTTP requests, sending WebSocket connections, and handling Server-Sent Events. Language bindings wrap this to provide native API surfaces.

Methods¶

graphql_at()¶

Send a GraphQL query/mutation to a custom endpoint

Signature:

def graphql_at(self, endpoint: str, query: str, variables: dict[str, Any], operation_name: str) -> ResponseSnapshot

graphql()¶

Send a GraphQL query/mutation

Signature:

def graphql(self, query: str, variables: dict[str, Any], operation_name: str) -> ResponseSnapshot

graphql_subscription_at()¶

Send a GraphQL subscription (WebSocket) to a custom endpoint.

Uses the graphql-transport-ws protocol and captures the first next payload. After the first payload is received, this client sends complete to unsubscribe.

Signature:

def graphql_subscription_at(self, endpoint: str, query: str, variables: dict[str, Any], operation_name: str) -> GraphQlSubscriptionSnapshot

graphql_subscription()¶

Send a GraphQL subscription (WebSocket).

Uses /graphql as the default subscription endpoint.

Signature:

def graphql_subscription(self, query: str, variables: dict[str, Any], operation_name: str) -> GraphQlSubscriptionSnapshot

TestingSseEvent¶

A single Server-Sent Event.

UploadFile¶

Represents an uploaded file from multipart/form-data requests.

This struct provides efficient access to file content with automatic base64 decoding and implements standard I/O traits for compatibility.

Methods¶

as_bytes()¶

Get the raw file content as bytes.

This provides zero-copy access to the underlying buffer.

Signature:

def as_bytes(self) -> bytes

read_to_string()¶

Read the file content as a UTF-8 string.

Errors:

Returns an error if the content is not valid UTF-8.

Signature:

def read_to_string(self) -> str

content_type_or_default()¶

Get the content type, defaulting to "application/octet-stream".

Signature:

def content_type_or_default(self) -> str

ValidateRequest¶

Request body for POST /asyncapi/validate

ValidationResponse¶

Response body for POST /asyncapi/validate

Enums¶

Method¶

HTTP method

SecuritySchemeInfo¶

Security scheme types

SnapshotError¶

Possible errors while converting an Axum response into a snapshot.

WebSocketMessage¶

A WebSocket message that can be text or binary.

Errors¶

AppError¶

Error type for application builder operations.

Base class: AppError(Exception)

GraphQlError¶

Errors that can occur during GraphQL operations

These errors are compatible with async-graphql error handling and can be converted to structured HTTP responses matching the project's error fixtures.

Base class: GraphQlError(Exception)

SchemaError¶

Error type for schema building operations

Base class: SchemaError(Exception)

Edit this page on GitHub