### Run Local Development Server Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/19-local-development.md Start the uvicorn server, pointing it to your application instance. This command assumes your app is named 'app' and is in a file named 'example'. ```console $ uvicorn example:app ``` -------------------------------- ### Run Basic Query Example Source: https://github.com/mirumee/ariadne/blob/main/CLAUDE.md Execute a basic GraphQL query example using the 'uv' command. Refer to the example file's docstring for specific instructions. ```bash # Run a basic query example uv run examples/basic_query_example.py ``` -------------------------------- ### Quick Start: SQLAlchemy Setup with Ariadne Source: https://github.com/mirumee/ariadne/blob/main/docs/07-Contrib/02-sqlalchemy.md Sets up a GraphQL schema with SQLAlchemy models, object types, and query types. It configures a synchronous SQLAlchemy session for use with WSGI and ASGI applications. The session is made available in the GraphQL context. ```python from ariadne import make_executable_schema from ariadne.asgi import GraphQL from ariadne.asgi.handlers import GraphQLHTTPHandler from ariadne.contrib.sqlalchemy import ( SQLAlchemyObjectType, SQLAlchemyQueryType, ) from sqlalchemy import Column, ForeignKey, Integer, String, Table, create_engine from sqlalchemy.orm import Session, declarative_base, relationship, sessionmaker Base = declarative_base() post_tags = Table( "post_tags", Base.metadata, Column("post_id", Integer, ForeignKey("posts.id"), primary_key=True), Column("tag_id", Integer, ForeignKey("tags.id"), primary_key=True), ) class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True) username = Column(String) posts = relationship("Post", back_populates="author") class Post(Base): __tablename__ = "posts" id = Column(Integer, primary_key=True) title = Column(String) author_id = Column(Integer, ForeignKey("users.id")) author = relationship("User", back_populates="posts") tags = relationship("Tag", secondary=post_tags, back_populates="posts") class Tag(Base): __tablename__ = "tags" id = Column(Integer, primary_key=True) name = Column(String) posts = relationship("Post", secondary=post_tags, back_populates="tags") type_defs = """ type Query { users: [User!]! posts: [Post!]! tags: [Tag!]! } type User { id: ID! username: String! posts: [Post!]! } type Post { id: ID! title: String! author: User! tags: [Tag!]! } type Tag { id: ID! name: String! posts: [Post!]! } """ user_type = SQLAlchemyObjectType("User", User) post_type = SQLAlchemyObjectType("Post", Post) tag_type = SQLAlchemyObjectType("Tag", Tag) query = SQLAlchemyQueryType([user_type, post_type, tag_type]) schema = make_executable_schema(type_defs, [query, user_type, post_type, tag_type]) engine = create_engine("sqlite:///db.sqlite3") SessionLocal = sessionmaker(engine, expire_on_commit=False) async def get_context(request, _data): return {"request": request, "session": SessionLocal()} app = GraphQL( schema, context_value=get_context, http_handler=GraphQLHTTPHandler(), ) ``` -------------------------------- ### SubscriptionType - Example Usage Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example demonstrating how to set up a SubscriptionType with source and field resolvers. ```APIDOC ## SubscriptionType Example ### Description This example shows how to define a subscription field with a source and a resolver, and then bind it to an executable GraphQL schema. ### Code Example ```python from ariadne import SubscriptionType, make_executable_schema from broadcast import broadcast from .models import Post subscription_type = SubscriptionType() @subscription_type.source("post") async def source_post(*_, category: str | None = None) -> dict: async with broadcast.subscribe(channel="NEW_POSTS") as subscriber: async for event in subscriber: message = json.loads(event.message) # Send message to resolver if we don't filter if not category or message["category"] == category: yield message @subscription_type.field("post") async def resolve_post( message: dict, *_, category: str | None = None ) -> Post: # Convert message to Post object that resolvers for Post type in # GraphQL schema understand. return await Post.get_one(id=message["post_id"]) schema = make_executable_schema( """ type Query { """Valid schema must define the Query type""" none: Int } type Subscription { post(category: ID): Post! } type Post { id: ID! author: String! text: String! } """, subscription_type ) ``` ``` -------------------------------- ### Dynamic Context Value Example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/06-types-reference.md An example of a dynamic context value callable that fetches settings and an authenticated user, then returns them along with the request object. This callable is evaluated at the start of query execution. ```python from ariadne.asgi import GraphQL from starlette.requests import Request from .auth import get_authenticated_user from .settings import get_db_settings from .schema import schema async def get_context_value(request: Request, _): settings = await get_db_settings() user = await get_authenticated_user(request, settings) return { "request": request, "settings": settings, "user": user, } graphql_app = GraphQL( schema, context_value=get_context_value, ) ``` -------------------------------- ### Install Broadcaster Library Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/12-subscriptions.md Installs the Broadcaster library, a pub-sub solution supporting Redis, PostgreSQL, and Apache Kafka, using pip. ```console pip install broadcaster ``` -------------------------------- ### Install ASGI Server Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/19-local-development.md Install uvicorn, an ASGI server, using pip. This is required to run the Ariadne GraphQL application. ```console $ pip install uvicorn ``` -------------------------------- ### Run ASGI Example with Uvicorn Source: https://github.com/mirumee/ariadne/blob/main/CLAUDE.md Run an ASGI application example using uvicorn, with specific dependencies like 'uvicorn[standard]' and 'ariadne'. The '--reload' flag enables hot-reloading during development. ```bash # Run an ASGI example with uvicorn uv run --with "uvicorn[standard]" --with ariadne \ uvicorn examples.basic_query_example:app --reload ``` -------------------------------- ### Install ASGI Server Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/01-intro.md Install uvicorn, an ASGI server, to run your Ariadne GraphQL application. ```bash pip install uvicorn ``` -------------------------------- ### Install Ariadne with File Uploads Support Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/14-file-uploads.md Install the necessary package for file upload functionality. This command installs Ariadne along with the 'python-multipart' library. ```console pip install "ariadne[file-uploads]" ``` -------------------------------- ### Subscription source and resolver example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Demonstrates defining a subscription source and resolver using SubscriptionType. ```python from ariadne import SubscriptionType, make_executable_schema from broadcast import broadcast from .models import Post subscription_type = SubscriptionType() @subscription_type.source("post") async def source_post(*_, category: str | None = None) -> dict: async with broadcast.subscribe(channel="NEW_POSTS") as subscriber: async for event in subscriber: message = json.loads(event.message) # Send message to resolver if we don't filter if not category or message["category"] == category: yield message @subscription_type.field("post") async def resolve_post( message: dict, *_, category: str | None = None ) -> Post: # Convert message to Post object that resolvers for Post type in # GraphQL schema understand. return await Post.get_one(id=message["post_id"]) schema = make_executable_schema( """ type Query { "Valid schema must define the Query type" none: Int } type Subscription { post(category: ID): Post! } type Post { id: ID! author: String! text: String! } """, subscription_type ) ``` -------------------------------- ### Install graphql-sync-dataloaders Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/13-dataloaders.md Install the synchronous dataloader library using pip. This is required for Python 3.8 and later. ```bash $ pip install graphql-sync-dataloaders ``` -------------------------------- ### Install Ariadne Source: https://github.com/mirumee/ariadne/blob/main/README.md Install the Ariadne library using pip. Ensure you have Python 3.10 or higher. ```console pip install ariadne ``` -------------------------------- ### Install aiodataloader Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/13-dataloaders.md Installs the aiodataloader package, required for asynchronous dataloader functionality. ```bash $ pip install aiodataloader ``` -------------------------------- ### Basic Flask GraphQL Server with Ariadne Source: https://github.com/mirumee/ariadne/blob/main/docs/05-Integrations/03-flask-integration.md This snippet sets up a complete GraphQL server using Ariadne within a Flask application. It includes schema definition, query resolution, GraphiQL explorer integration, and request handling for both GET and POST methods. Ensure Flask and Ariadne are installed. ```python from ariadne import QueryType, graphql_sync, make_executable_schema from ariadne.explorer import ExplorerGraphiQL from flask import Flask, jsonify, request type_defs = """ type Query { hello: String! } """ query = QueryType() @query.field("hello") def resolve_hello(_, info): request = info.context user_agent = request.headers.get("User-Agent", "Guest") return "Hello, %s!" % user_agent schema = make_executable_schema(type_defs, query) app = Flask(__name__) # Retrieve HTML for the GraphiQL. # If explorer implements logic dependant on current request, # change the html(None) call to the html(request) # and move this line to the graphql_explorer function. explorer_html = ExplorerGraphiQL().html(None) @app.route("/graphql", methods=["GET"]) def graphql_explorer(): # On GET request serve the GraphQL explorer. # You don't have to provide the explorer if you don't want to # but keep on mind this will not prohibit clients from # exploring your API using desktop GraphQL explorer app. return explorer_html, 200 @app.route("/graphql", methods=["POST"]) def graphql_server(): # GraphQL queries are always sent as POST data = request.get_json() # Note: Passing the request to the context is optional. # In Flask, the current request is always accessible as flask.request success, result = graphql_sync( schema, data, context_value={"request": request}, debug=app.debug ) status_code = 200 if success else 400 return jsonify(result), status_code if __name__ == "__main__": app.run(debug=True) ``` -------------------------------- ### HTTP Callback Handler Example Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/24-subscription-handlers.md Demonstrates implementing a subscription handler for gateway architectures using HTTP callbacks. ```APIDOC ## HTTP Callback Handler Example This example shows how to implement a subscription handler for gateway architectures where events are delivered via HTTP callbacks instead of persistent connections. ### Use Case ``` Client ──SSE──> Gateway ──HTTP POST──> GraphQL Server (DGS) ↑ │ └────HTTP callbacks────────┘ ``` 1. Client connects to gateway via SSE 2. Gateway forwards subscription to DGS with a `callbackUri` 3. DGS returns 204 immediately 4. DGS posts events (DATA, KEEP_ALIVE, COMPLETE, CLOSE) to the callback URI 5. Gateway streams events to client ``` -------------------------------- ### Example GraphQL query Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/02-resolvers.md A sample query that triggers the execution of multiple resolvers. ```graphql { user { username } } ``` -------------------------------- ### Callback Payload Examples Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/24-subscription-handlers.md Examples of JSON payloads for different subscription event types sent via HTTP callbacks. ```APIDOC ### Callback Payloads **KEEP_ALIVE:** ```json { "subscriptionId": "uuid-here", "type": "KEEP_ALIVE" } ``` **DATA:** ```json { "subscriptionId": "uuid-here", "type": "DATA", "data": { "counter": 42 } } ``` **COMPLETE:** ```json { "subscriptionId": "uuid-here", "type": "COMPLETE" } ``` **CLOSE (with error):** ```json { "subscriptionId": "uuid-here", "type": "CLOSE", "errors": [{"message": "Subscription error: ..."}] } ``` ``` -------------------------------- ### Install Uvicorn ASGI Server Source: https://github.com/mirumee/ariadne/blob/main/README.md Install uvicorn, an ASGI server, to run your Ariadne application. This is required for serving the API locally. ```console pip install uvicorn ``` -------------------------------- ### Example with Directive Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Demonstrates how to use custom schema directives in Ariadne. This example defines an @uppercase directive to convert string results to uppercase. ```python from functools import wraps from ariadne import SchemaDirectiveVisitor, graphql_sync, make_executable_schema from graphql import default_field_resolver class UppercaseDirective(SchemaDirectiveVisitor): def visit_field_definition(self, field, object_type): org_resolver = field.resolve or default_field_resolver @wraps(org_resolver) def uppercase_resolved_value(*args, **kwargs): value = org_resolver(*args, **kwargs) if isinstance(value, str): return value.upper() return value # Extend field's behavior by wrapping it's resolver in custom one field.resolve = uppercase_resolved_value return field schema = make_executable_schema( """ directive @uppercase on FIELD_DEFINITION type Query { helloWorld: String! @uppercase } """, directives={"uppercase": UppercaseDirective}, ) no_errors, result = graphql_sync( schema, {"query": "{ helloWorld }"}, root_value={"helloWorld": "Hello world!"}, ) assert no_errors assert result == { "data": { "helloWorld": "HELLO WORLD!", }, } ``` -------------------------------- ### Run All Checks with Just Source: https://github.com/mirumee/ariadne/blob/main/CONTRIBUTING.md Installs 'just' and runs all CI checks including formatting, type checking, and tests. Use this before submitting a pull request. ```bash just check ``` -------------------------------- ### Full implementation using lambda for deserialization Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/04-inputs.md A complete example showing the schema definition, dataclass, resolver, and InputType configuration using a lambda function. ```python from dataclasses import dataclass from typing import List, Optional from ariadne import InputType, MutationType, gql, make_executable_schema type_defs = gql( """ type Query { unused: Boolean } type Mutation { issueCreate( input: IssueInput! ): IssueMutationResult! } input IssueInput { title: String!, description: String! labels: [String!]! priority: Int! isClosed: Boolean! } type IssueMutationResult { success: Boolean! error: String! } """ ) @dataclass class IssueInput: title: str description: str labels: List[str] priority: int is_closed: Optional[bool] = None mutation_type = MutationType() @mutation_type.field("issueCreate") def resolve_issue_create(_, info, input: IssueInput): try: create_new_issue(info.context, input) return {"success": True} except ValidationError as err: return { "success": False, "error": str(err), } schema = make_executable_schema( type_defs, mutation_type, InputType( "IssueInput", lambda data: IssueInput(**data), {"isClosed": "is_closed"}, ), ) ``` -------------------------------- ### Install Ariadne with SQLAlchemy Extra Source: https://github.com/mirumee/ariadne/blob/main/docs/07-Contrib/02-sqlalchemy.md Install Ariadne with the sqlalchemy extra to include SQLAlchemy and aiodataloader dependencies. ```console pip install ariadne[sqlalchemy] ``` -------------------------------- ### Simple GraphQL Chat Example Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/12-subscriptions.md A complete example of a GraphQL chat application using Ariadne, implementing mutations for sending messages and subscriptions for receiving them. It utilizes `broadcaster` for message publishing and subscription. ```python import json from ariadne import MutationType, SubscriptionType, make_executable_schema from ariadne.asgi import GraphQL from ariadne.asgi.handlers import GraphQLWSHandler from broadcaster import Broadcast from starlette.applications import Starlette broadcast = Broadcast("memory://") type_defs = """ type Query { _unused: Boolean } type Message { sender: String message: String } type Mutation { send(sender: String!, message: String!): Boolean } type Subscription { message: Message } """ mutation = MutationType() @mutation.field("send") async def resolve_send(*_, **message): await broadcast.publish(channel="chatroom", message=json.dumps(message)) return True subscription = SubscriptionType() @subscription.source("message") async def source_message(_, info): async with broadcast.subscribe(channel="chatroom") as subscriber: async for event in subscriber: yield json.loads(event.message) schema = make_executable_schema(type_defs, mutation, subscription) graphql = GraphQL( schema=schema, debug=True, websocket_handler=GraphQLWSHandler(), ) app = Starlette( debug=True, on_startup=[broadcast.connect], on_shutdown=[broadcast.disconnect], ) app.mount("/", graphql) ``` -------------------------------- ### GraphQL Query Example Source: https://github.com/mirumee/ariadne/blob/main/docs/06-Extensions/02-middleware.md A sample GraphQL query used to illustrate how middleware execution scales with the number of fields returned. ```graphql { users { id email username } } ``` -------------------------------- ### Implicit type resolution examples Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Examples of using __typename for implicit type resolution. ```python user_data_dict = {"__typename": "User", ...} # or... class UserRepr: __typename: str = "User" ``` -------------------------------- ### Asynchronous Lambda Handler Example Source: https://github.com/mirumee/ariadne/blob/main/docs/04-Servers/03-aws-lambda.md This example demonstrates an asynchronous Lambda handler using `graphql` and the `async_to_sync` decorator for running in an event loop. It requires `asgiref`. ```python import json import logging from ariadne import QueryType, graphql, make_executable_schema, gql from asgiref.sync import async_to_sync logger = logging.getLogger() type_defs = gql( """ type Query { hello: String! } """ ) query_type = QueryType() @query_type.field("hello") def resolve_hello(_, info): http_context = info.context["requestContext"]["http"] user_agent = http_context.get("userAgent") or "Anon" return f"Hello {user_agent}!" schema = make_executable_schema(type_defs, query_type) @async_to_sync async def handler(event: dict, _): try: data = json.loads(event.get("body") or "") except ValueError as exc: return response({"error": f"Failed to parse JSON: {exc}"}}, 405) success, result = await graphql( schema, data, context_value=event, logger=logger, ) return response(result, 200 if success else 400) def response(body: dict, status_code: int = 200): return { "statusCode": status_code, "headers": { "Content-Type": "application/json" }, "body": json.dumps(body), } ``` -------------------------------- ### Use combine_multipart_data for file uploads Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example demonstrating how to populate an operation's variables with file data using a mapping dictionary. ```python # Single GraphQL operation operations = { "operationName": "AvatarUpload", "query": """ mutation AvatarUpload($type: String!, $image: Upload!) { avatarUpload(type: $type, image: $image) { success errors } } """, "variables": {"type": "SQUARE", "image": None} } files_map = {"0": ["variables.image"]} files = {"0": UploadedFile(....)} combine_multipart_data(operations, files_map, files assert operations == { "variables": {"type": "SQUARE", "image": UploadedFile(....)} } ``` -------------------------------- ### Start WebSocket operation Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/03-asgi-handlers-reference.md Initiates a new GraphQL operation over the WebSocket. ```python async def start_websocket_operation( self, websocket: WebSocket, data: Any, context_value: Any, query_document: DocumentNode, operation_id: str, operations: dict[str, Operation], ) -> None: ... ``` -------------------------------- ### Create minimal executable schema Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md An example of creating a schema without Python logic, relying on root_value for query execution. ```python from ariadne import graphql_sync, make_executable_schema schema = make_executable_schema( """ type Query { helloWorld: String! } """ ) no_errors, result = graphql_sync( schema, {"query": "{ helloWorld }"}, root_value={"helloWorld": "Hello world!"}, ) assert no_errors assert result == { "data": { "helloWorld": "Hello world!", }, } ``` -------------------------------- ### Initialize Ariadne with subscriptions-transport-ws Handler Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/12-subscriptions.md Use this handler for the default `subscriptions-transport-ws` protocol. Ensure you have `websockets` installed if using uvicorn. ```python from ariadne.asgi import GraphQL from ariadne.asgi.handlers import GraphQLWSHandler graphql_app = GraphQL( schema, websocket_handler=GraphQLWSHandler(), ) ``` -------------------------------- ### HTTP Callback Payloads Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/24-subscription-handlers.md Example JSON payloads for different subscription callback events. ```json { "subscriptionId": "uuid-here", "type": "KEEP_ALIVE" } ``` ```json { "subscriptionId": "uuid-here", "type": "DATA", "data": { "counter": 42 } } ``` ```json { "subscriptionId": "uuid-here", "type": "COMPLETE" } ``` ```json { "subscriptionId": "uuid-here", "type": "CLOSE", "errors": [{"message": "Subscription error: ..."}] } ``` -------------------------------- ### Example with Converted Names Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Shows how to use `convert_names_case=True` to map camelCase GraphQL field names to snake_case Python dictionary keys. ```python from ariadne import graphql_sync, make_executable_schema schema = make_executable_schema( """ type Query { helloWorld: String! } """, convert_names_case=True, ) no_errors, result = graphql_sync( schema, {"query": "{ helloWorld }"}, root_value={"hello_world": "Hello world!"}, ) assert no_errors assert result == { "data": { "helloWorld": "Hello world!", }, } ``` -------------------------------- ### Implement custom QueryParser Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/06-types-reference.md Example of a custom parser that restricts queries to a predefined set based on operation name. ```python from graphql import GraphQLError, parse class AllowedQueriesParser: def __init__(self): self._queries = {} def __call__(self, _, data): operation_name = data.get("operationName") if not operation_name: raise GraphQLError( "Explicit 'operationName' is required by the server." ) if operation_name not in self._queries: raise GraphQLError( f"Operation 'operation_name' is not supported by the server." ) return self._queries[operation_name] def add_query(self, operation_name: str, query: str): self._queries[operation_name] = parse(query) allowed_queries_parser = AllowedQueriesParser() allowed_queries_parser.add_query( "GetUsers", """ query GetUsers { users { id name } } """ ) ``` -------------------------------- ### Initialize GraphQL ASGI application Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/02-asgi-reference.md Constructor for configuring the GraphQL server, including schema, handlers, and execution settings. ```python def __init__( self, schema: GraphQLSchema, *, context_value: ContextValue | None = None, root_value: RootValue | None = None, query_parser: QueryParser | None = None, query_validator: QueryValidator | None = None, validation_rules: ValidationRules | None = None, execute_get_queries: bool = False, debug: bool = False, introspection: bool = True, explorer: Explorer | None = None, logger: None | str | Logger | LoggerAdapter = None, error_formatter: ErrorFormatter = format_error, execution_context_class: type[ExecutionContext] | None = None, http_handler: GraphQLHTTPHandler | None = None, websocket_handler: GraphQLWebsocketHandlerBase | None = None, ) -> None: ... ``` -------------------------------- ### Callback Subscription Handler Usage Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/24-subscription-handlers.md Example of how to integrate a custom callback subscription handler into the Ariadne ASGI application. ```APIDOC ### Usage ```python from ariadne.asgi import GraphQL from ariadne.asgi.handlers import GraphQLHTTPHandler from myapp.handlers import CallbackSubscriptionHandler app = GraphQL( schema, http_handler=GraphQLHTTPHandler( subscription_handlers=[ CallbackSubscriptionHandler( keep_alive_interval=5.0, callback_timeout=10.0, max_retries=3, ) ] ), ) ``` ``` -------------------------------- ### Subscription Event Construction and Access Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/24-subscription-handlers.md Demonstrates how to instantiate subscription events and access their properties. ```python from ariadne.subscription_handlers.events import ( SubscriptionEvent, SubscriptionEventType, ) from graphql import ExecutionResult # Create events using direct construction event = SubscriptionEvent( event_type=SubscriptionEventType.NEXT, result=execution_result, ) event = SubscriptionEvent( event_type=SubscriptionEventType.ERROR, result=execution_result, ) event = SubscriptionEvent(event_type=SubscriptionEventType.COMPLETE) event = SubscriptionEvent(event_type=SubscriptionEventType.KEEP_ALIVE) # Access event data event.event_type # SubscriptionEventType event.result # ExecutionResult or None ``` -------------------------------- ### Handle GET Request Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Processes a WSGI HTTP GET request. ```python def handle_get(self, environ: dict, start_response) -> list[bytes]: ... ``` -------------------------------- ### Using SSESubscriptionHandler with GraphQLHTTPHandler Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/12-subscriptions.md This example demonstrates how to integrate SSESubscriptionHandler into your ASGI application to enable GraphQL subscriptions over Server-Sent Events. ```APIDOC ## Using Server-Sent Events Protocol Instead of WebSockets, [Server-Sent Events (SSE)](https://github.com/enisdenjo/graphql-sse/blob/master/PROTOCOL.md) can be used as a transmission protocol for subscriptions. This approach uses single, long-lived HTTP connection to push data to clients. To enable subscriptions over Server-Sent Events, use `SSESubscriptionHandler` with `GraphQLHTTPHandler`: ```python import asyncio from typing import Any, AsyncGenerator from ariadne import SubscriptionType, make_executable_schema, gql from ariadne.asgi import GraphQL from ariadne.asgi.handlers import GraphQLHTTPHandler from ariadne.contrib.sse import SSESubscriptionHandler type_defs = gql(""" type Query { _empty: String } type Subscription { counter: Int! } """) subscription = SubscriptionType() @subscription.field("counter") def counter_resolver(count: int, info: Any) -> int: return count @subscription.source("counter") async def counter_generator(obj: Any, info: Any) -> AsyncGenerator[int, None]: for i in range(5): yield i await asyncio.sleep(1) schema = make_executable_schema(type_defs, [subscription]) app = GraphQL( schema, http_handler=GraphQLHTTPHandler( subscription_handlers=[SSESubscriptionHandler()], ), ) ``` Subscriptions can be consumed using the [graphql-sse](https://github.com/enisdenjo/graphql-sse/) JavaScript client library or any other compatible implementation. > The `SSESubscriptionHandler` requires the ASGI server to work. > > This handler only supports the _distinct connections_ mode of the protocol due to Ariadne's stateless implementation. > > If you are using a custom client implementation, make sure to add the `Accept: text/event-stream` header to the request as this is required to establish the Server-Sent Events connection. ### SSESubscriptionHandler options The `SSESubscriptionHandler` accepts the following optional arguments: - `send_timeout`: the timeout in seconds to send an event to the client. - `ping_interval`: the interval in seconds to send a ping event to the client (defaults to 15 seconds). - `default_response_headers`: a dictionary of additional headers to be sent with the SSE response. ### Deprecated: GraphQLHTTPSSEHandler The `GraphQLHTTPSSEHandler` from `ariadne.contrib.sse` is deprecated. Use `SSESubscriptionHandler` with `GraphQLHTTPHandler` instead, as shown above. --- ``` -------------------------------- ### Handle GET Explorer Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Processes a WSGI HTTP GET explorer request. ```python def handle_get_explorer(self, environ: dict, start_response) -> list[bytes]: ... ``` -------------------------------- ### Initialize QueryType Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Two equivalent ways to initialize a Query type. ```python query_type = QueryType() ``` ```python query_type = ObjectType("Query") ``` -------------------------------- ### Handle GET Query Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Processes a WSGI HTTP GET query request. ```python def handle_get_query( self, environ: dict, start_response, query_params: dict, ) -> list[bytes]: ... ``` -------------------------------- ### Run Ariadne Development Server Source: https://github.com/mirumee/ariadne/blob/main/README.md Start the uvicorn development server to run the Ariadne GraphQL API. The server will be accessible at http://127.0.0.1:8000. ```console uvicorn example:app ``` -------------------------------- ### Manual Context Wiring with LoaderRegistry Source: https://github.com/mirumee/ariadne/blob/main/docs/07-Contrib/02-sqlalchemy.md An alternative to using SQLAlchemyDataLoaderExtension is to manually wire the LoaderRegistry in the get_context function. This provides explicit control over context setup. ```python from ariadne.contrib.sqlalchemy import LoaderRegistry async def get_context(request, _data): session = request.state.session return { "request": request, "session": session, "loader_registry": LoaderRegistry(session), } ``` -------------------------------- ### Initialize GraphQL Playground Source: https://github.com/mirumee/ariadne/blob/main/ariadne/explorer/templates/playground.html Initialize the GraphQL Playground by providing the target DOM element and configuration options. Ensure the DOM element is available before initialization. ```javascript window.addEventListener('load', function (event) { GraphQLPlayground.init(document.getElementById('root'), { // options as 'endpoint' belong here {% if settings %}settings: {% raw settings %},{% endif %} {% if share_enabled %}shareEnabled: {% raw share_enabled %},{% endif %} }) }) ``` -------------------------------- ### Serialize Date Example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example of using the @date_scalar.serializer decorator to serialize dates as 'YYYY-MM-DD' strings. ```python date_scalar = ScalarType("Date") @date_scalar.serializer def serialize_date(value: date) -> str: # Serialize dates as "YYYY-MM-DD" string return date.strftime("%Y-%m-%d") ``` -------------------------------- ### Initialize Extension Manager Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Initializes extensions and stores them with context on the instance. Optional arguments include a list of Extension types and context value. ```python def __init__( self, extensions: ExtensionList | None = None, context: ContextValue | None = None, ) -> None: ... ``` -------------------------------- ### Basic Synchronous Generator Subscription Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/23-sync-subscriptions.md Defines a basic synchronous generator for a subscription. This example demonstrates yielding messages from a blocking database call without needing async/await. ```python from ariadne import SubscriptionType, make_executable_schema subscription = SubscriptionType() @subscription.source("messages") def message_source(*_, channel: str = "default"): # This is a synchronous generator - no async/await needed! for message in get_messages_from_database(channel): # Blocking DB call yield {"text": message.text, "author": message.author} schema = make_executable_schema( """ type Query { _: Boolean } type Subscription { messages(channel: String): Message! } type Message { text: String! author: String! } """, subscription, ) ``` -------------------------------- ### Create Executable Schema with Multiple Bindables Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/01-intro.md Demonstrates passing a list of bindables (e.g., query, user, mutations) to make_executable_schema for a more complex API. ```python make_executable_schema(type_defs, [query, user, mutations]) ``` -------------------------------- ### Extract Data from GET Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Extracts unvalidated GraphQL query data from a GET request's querystring. ```python def extract_data_from_get(self, query_params: dict) -> dict: ... ``` -------------------------------- ### Extract GraphQL Data from GET Request Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/03-asgi-handlers-reference.md Extracts unvalidated GraphQL query data from the querystring of a GET request. ```python def extract_data_from_get_request(self, request: Request) -> dict: ... ``` -------------------------------- ### GraphiQL Initialization Script Source: https://github.com/mirumee/ariadne/blob/main/ariadne/explorer/templates/graphiql.html JavaScript logic to initialize the GraphiQL fetcher, plugins, and render the React component. ```javascript import React from 'react'; import ReactDOM from 'react-dom/client'; import { GraphiQL } from 'graphiql'; import { createGraphiQLFetcher } from '@graphiql/toolkit'; {% if enable\_explorer\_plugin %} import { explorerPlugin } from '@graphiql/plugin-explorer'; {% endif %} import 'graphiql/setup-workers/esm.sh'; const fetcher = createGraphiQLFetcher({ url: window.location.href, {% if subscription\_url %} subscriptionUrl: "{{ subscription\_url }}", {% else %} subscriptionUrl: (window.location.protocol === "https:" ? "wss" : "ws") + "://" + window.location.host + window.location.pathname, {% endif %} }); {% if enable\_explorer\_plugin %} const plugins = [explorerPlugin()]; {% endif %} function AriadneGraphiQL() { return React.createElement(GraphiQL, { fetcher, {% if enable\_explorer\_plugin %} plugins, {% endif %} defaultEditorToolsVisibility: true, initialQuery: '{% raw default\_query %}', }); } const container = document.getElementById('graphiql'); const root = ReactDOM.createRoot(container); root.render(React.createElement(AriadneGraphiQL)); ``` -------------------------------- ### Union type resolver example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example of a type resolver function that maps Python objects to GraphQL type names. ```python def example_type_resolver(obj: Any, *_) -> str: if isinstance(obj, PythonReprOfUser): return "USer" if isinstance(obj, PythonReprOfComment): return "Comment" raise ValueError(f"Don't know GraphQL type for '{obj}'!") ``` -------------------------------- ### Instantiate Broadcaster with Redis Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/12-subscriptions.md Initializes a Broadcaster instance for Redis, connecting to a local Redis server on the default port. ```python broadcast = Broadcast("redis://localhost:6379") ``` -------------------------------- ### Implement ASGI call method Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/02-asgi-reference.md The entrypoint for the ASGI application, supporting both HTTP and WebSocket connections. ```python async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ... ``` -------------------------------- ### Example Datetime Scalar Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md This example demonstrates how to define a custom DateTime scalar in Ariadne, including serialization and value parsing. ```APIDOC ## Example Datetime Scalar This code defines a datetime scalar that converts Python datetime objects to and from strings. It includes custom serializer and value parser functions. ```python from datetime import datetime from ariadne import QueryType, ScalarType, make_executable_schema scalar_type = ScalarType("DateTime") @scalar_type.serializer def serialize_value(val: datetime) -> str: return datetime.strftime(val, "%Y-%m-%d %H:%M:%S") @scalar_type.value_parser def parse_value(val) -> datetime: if not isinstance(val, str): raise ValueError( f"'{val}' is not a valid JSON representation " ) return datetime.strptime(val, "%Y-%m-%d %H:%M:%S") query_type = QueryType() @query_type.field("now") def resolve_now(*_): return datetime.now() @query_type.field("diff") def resolve_diff(*_, value): delta = datetime.now() - value return int(delta.total_seconds()) schema = make_executable_schema( """ scalar DateTime type Query { now: DateTime! diff(value: DateTime): Int! } """, scalar_type, query_type, ) ``` ``` -------------------------------- ### Configure Ariadne GraphQL App Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/19-local-development.md Create an instance of ariadne.asgi.GraphQL, passing in an executable schema. This app will be served by an ASGI server. ```python from ariadne import make_executable_schema from ariadne.asgi import GraphQL from . import type_defs, resolvers schema = make_executable_schema(type_defs, resolvers) app = GraphQL(schema) ``` -------------------------------- ### Run Quick Development Checks Source: https://github.com/mirumee/ariadne/blob/main/CLAUDE.md Execute format, type checking, and tests (excluding integration tests) with a single command. This is useful for quick local checks before committing. ```bash # Quick checks: format + types + tests (no integration tests) just quick-check ``` -------------------------------- ### GraphQLMiddleware WSGI Constructor Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Initializes the GraphQLMiddleware for WSGI applications. It routes requests to either a main application or a specific GraphQL application based on the path. ```python class GraphQLMiddleware: ... def __init__( self, app: Callable, graphql_app: Callable, path: str = '/graphql/', ): ... ``` -------------------------------- ### InterfaceType Example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md An example demonstrating the usage of InterfaceType to create a schema with a field returning different GraphQL types based on the resolved object. ```APIDOC ### Example Following code creates a [GraphQL schema](https://graphql-core-3.readthedocs.io/en/latest/modules/type.html#graphql.type.GraphQLSchema) with a field that returns random result of either `User` or `Post` GraphQL type. It also supports dict with `__typename` key that explicitly declares its GraphQL type: ```python import random from dataclasses import dataclass from ariadne import QueryType, InterfaceType, make_executable_schema @dataclass class UserModel: id: str name: str @dataclass class PostModel: id: str message: str results = ( UserModel(id=1, name="Bob"), UserModel(id=2, name="Alice"), UserModel(id=3, name="Jon"), PostModel(id=1, message="Hello world!"), PostModel(id=2, message="How's going?"), PostModel(id=3, message="Sure thing!"), {"__typename": "User", "id": 4, "name": "Polito"}, {"__typename": "User", "id": 5, "name": "Aerith"}, {"__typename": "Post", "id": 4, "message": "Good day!"}, {"__typename": "Post", "id": 5, "message": "Whats up?"}, ) query_type = QueryType() @query_type.field("result") def resolve_random_result(*_): return random.choice(results) result_type = InterfaceType("Result") @result_type.type_resolver def resolve_result_type(obj: UserModel | PostModel | dict, *_) -> str: if isinstance(obj, UserModel): return "User" if isinstance(obj, PostModel): return "Post" if isinstance(obj, dict) and obj.get("__typename"): return obj["__typename"] raise ValueError(f"Don't know GraphQL type for '{obj}'!") schema = make_executable_schema( """ type Query { result: Result! } interface Result { id: ID! } type User implements Result { id: ID! name: String! } type Post implements Result { id: ID! message: String! } """, query_type, result_type, ) ``` ``` -------------------------------- ### Define DateTime Scalar Example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example defining a DateTime scalar that converts Python datetime objects to and from strings. Includes serializer and value parser functions. ```python from datetime import datetime from ariadne import QueryType, ScalarType, make_executable_schema scalar_type = ScalarType("DateTime") @scalar_type.serializer def serialize_value(val: datetime) -> str: return datetime.strftime(val, "%Y-%m-%d %H:%M:%S") @scalar_type.value_parser def parse_value(val) -> datetime: if not isinstance(val, str): raise ValueError( f"'{{val}}' is not a valid JSON representation " ) return datetime.strptime(val, "%Y-%m-%d %H:%M:%S") query_type = QueryType() @query_type.field("now") def resolve_now(*_): return datetime.now() @query_type.field("diff") def resolve_diff(*_, value): delta = datetime.now() - value return int(delta.total_seconds()) schema = make_executable_schema( """ scalar DateTime type Query { now: DateTime! diff(value: DateTime): Int! } """, scalar_type, query_type, ) ``` -------------------------------- ### Parse Date String Example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example of using the @date_scalar.value_parser decorator to parse 'YYYY-MM-DD' strings into date objects. Includes error handling for invalid formats. ```python date_scalar = ScalarType("Date") @date_scalar.value_parser def parse_date_str(value: str) -> date: try: # Parse "YYYY-MM-DD" string into date return datetime.strptime(value, "%Y-%m-%d").date() except (ValueError, TypeError): raise ValueError( f'"{{value}}" is not a date string in YYYY-MM-DD format.' ) ``` -------------------------------- ### Create Ariadne ASGI Application Source: https://github.com/mirumee/ariadne/blob/main/docs/01-Docs/01-intro.md Instantiate a `ariadne.asgi.GraphQL` application with your schema and enable debug mode. ```python from ariadne.asgi import GraphQL app = GraphQL(schema, debug=True) ``` -------------------------------- ### handle Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/03-asgi-handlers-reference.md Entrypoint for the GraphQL WebSocket handler called by the Ariadne ASGI application. ```APIDOC ## handle ### Description Entrypoint for the GraphQL WebSocket handler, called by the Ariadne ASGI GraphQL application to handle incoming connections. ### Parameters - **scope** (dict) - Required - Connection scope information. - **receive** (callable) - Required - Awaitable callable yielding event dictionaries. - **send** (callable) - Required - Awaitable callable for sending event dictionaries. ``` -------------------------------- ### Parse Date Literal Example Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example of using the @date_scalar.literal_parser decorator to parse date literals. It expects a StringValueNode and converts it to a date object, with error handling for invalid formats. ```python date_scalar = ScalarType("Date") @date_scalar.literal_parser def parse_date_literal( value: str, variable_values: dict[str, Any] | None = None ) -> date: if not isinstance(ast, StringValueNode): raise ValueError() try: # Parse "YYYY-MM-DD" string into date return datetime.strptime(ast.value, "%Y-%m-%d").date() except (ValueError, TypeError): raise ValueError( f'"{{value}}" is not a date string in YYYY-MM-DD format.' ) ``` -------------------------------- ### Implement Scalar serialization Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/01-api-reference.md Example of a serializer function for a date scalar. ```python def serialize_date(value: date) -> str: # Serialize dates as "YYYY-MM-DD" string return date.strftime("%Y-%m-%d") ``` -------------------------------- ### extract_data_from_get Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Extracts GraphQL data from GET request's querystring. ```APIDOC ## extract_data_from_get ### Description Extracts GraphQL data from GET request's querystring. Returns a `dict` with GraphQL query data that was not yet validated. ### Method GET ### Endpoint N/A (Internal function) ### Parameters #### Required arguments - **query_params** (dict) - A `dict` with parsed query string. ``` -------------------------------- ### Run Benchmarks Source: https://github.com/mirumee/ariadne/blob/main/CLAUDE.md Execute performance benchmarks and store the results. The '--benchmark-storage' option specifies where to save the results, here using a file-based backend. ```bash # Run benchmarks hatch test benchmark --benchmark-storage=file://benchmark/results ``` -------------------------------- ### handle_get_query Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Handles WSGI HTTP GET request with query parameters. ```APIDOC ## handle_get_query ### Description Handles WSGI HTTP GET request with query parameters. ### Method GET ### Endpoint N/A (Internal function) ### Parameters #### Required arguments - **environ** (dict) - A WSGI environment dictionary. - **start_response** (Callable) - A callable used to begin new HTTP response. - **query_params** (dict) - A dictionary of query parameters. ``` -------------------------------- ### Custom Extension with Sync/Async Resolver Handling Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/06-types-reference.md Implement a custom extension that correctly handles both synchronous and asynchronous resolvers. This example measures the execution time of resolvers and stores path-time pairs. ```python from inspect import iscoroutinefunction from time import time from ariadne.types import Extension, Resolver from graphql import GraphQLResolveInfo from graphql.pyutils import is_awaitable class MyExtension(Extension): def __init__(self): self.paths = [] def resolve( self, next_: Resolver, obj: Any, info: GraphQLResolveInfo, **kwargs ) -> Any: path = ".".join(map(str, info.path.as_list())) # Fast implementation for synchronous resolvers if not iscoroutinefunction(next_): start_time = time() result = next_(obj, info, **kwargs) self.paths.append((path, time() - start_time)) return result # Create async closure for async `next_` that GraphQL # query executor will handle for us. async def async_my_extension(): start_time = time() result = await next_(obj, info, **kwargs) if is_awaitable(result): result = await result self.paths.append((path, time() - start_time)) return result # GraphQL query executor will execute this closure for us return async_my_extension() ``` -------------------------------- ### handle_get Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Handles WSGI HTTP GET request and returns a response to the client. ```APIDOC ## handle_get ### Description Handles WSGI HTTP GET request and returns a response to the client. Returns list of bytes with response body. ### Method GET ### Endpoint N/A (Internal function) ### Parameters #### Required arguments - **environ** (dict) - A WSGI environment dictionary. - **start_response** (Callable) - A callable used to begin new HTTP response. ``` -------------------------------- ### Enable Extensions with a List Source: https://github.com/mirumee/ariadne/blob/main/docs/06-Extensions/01-extensions.md Initialize the GraphQL handler with a list of extension classes to enable them for all requests. ```python from ariadne.asgi import GraphQL from ariadne.asgi.handlers import GraphQLHTTPHandler from .schema import schema from .extensions import MyExtension app = GraphQL( schema, http_handler=GraphQLHTTPHandler( extensions=[MyExtension], ), ) ``` -------------------------------- ### Get Request Data Source: https://github.com/mirumee/ariadne/blob/main/docs/08-API-reference/07-wsgi-reference.md Extracts unvalidated GraphQL request data from the environment. ```python def get_request_data(self, environ: dict) -> Any: ... ```