GraphQL client for Python that enables developers to execute GraphQL queries, mutations, and subscriptions using multiple transport protocols including HTTP, WebSockets, and local schemas with support for both synchronous and asynchronous usage patterns
npx @tessl/cli install tessl/pypi-gql@4.0.0A comprehensive GraphQL client library for Python that enables developers to execute GraphQL queries, mutations, and subscriptions using multiple transport protocols including HTTP, WebSockets (with support for Apollo, GraphQL-WS, Phoenix channels, and AWS AppSync), and file-based operations. The library offers both synchronous and asynchronous usage patterns with support for concurrent requests, local query validation using GraphQL schemas, file uploads, custom scalars and enums, request batching, and a command-line interface for executing queries and downloading schemas.
pip install gqlfrom gql import gql, Client, GraphQLRequest, FileVarFor transport classes:
from gql.transport.requests import RequestsHTTPTransport
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.httpx import HTTPXTransport, HTTPXAsyncTransport
from gql.transport.websockets import WebsocketsTransportFor DSL (Domain Specific Language) functionality:
from gql.dsl import DSLSchema, DSLQuery, dsl_gqlfrom gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
# Create transport and client
transport = RequestsHTTPTransport(url="https://api.example.com/graphql")
client = Client(transport=transport, fetch_schema_from_transport=True)
# Execute a query
query = gql('''
query GetUser($id: ID!) {
user(id: $id) {
name
email
}
}
''')
result = client.execute(query, variable_values={"id": "123"})
print(result["user"]["name"])import asyncio
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport
async def main():
transport = AIOHTTPTransport(url="https://api.example.com/graphql")
async with Client(transport=transport) as session:
query = gql('{ hello }')
result = await session.execute(query)
print(result)
asyncio.run(main())import asyncio
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport
async def main():
transport = WebsocketsTransport(url="wss://api.example.com/graphql")
async with Client(transport=transport) as session:
subscription = gql('''
subscription {
messageAdded {
id
content
user
}
}
''')
async for result in session.subscribe(subscription):
print(f"New message: {result}")
asyncio.run(main())The gql library follows a modular architecture with clear separation of concerns:
This design enables flexible transport switching, comprehensive async/sync support, and seamless integration with the broader GraphQL ecosystem including graphene, graphql-core, and graphql-js.
Core client functionality for executing GraphQL operations, managing connections, and handling schemas. Includes both synchronous and asynchronous clients with session management for connection pooling and batched operations.
class Client:
def __init__(
self,
*,
schema: Optional[Union[str, GraphQLSchema]] = None,
introspection: Optional[IntrospectionQuery] = None,
transport: Optional[Union[Transport, AsyncTransport]] = None,
fetch_schema_from_transport: bool = False,
introspection_args: Optional[Dict] = None,
execute_timeout: Optional[Union[int, float]] = 10,
serialize_variables: bool = False,
parse_results: bool = False,
batch_interval: float = 0,
batch_max: int = 10
): ...
def execute(self, request: GraphQLRequest, **kwargs) -> ExecutionResult: ...
async def execute_async(self, request: GraphQLRequest, **kwargs) -> ExecutionResult: ...
def session(self, **kwargs) -> SyncClientSession: ...
async def connect_async(self, **kwargs) -> AsyncClientSession: ...
def gql(request_string: str) -> GraphQLRequest: ...
class GraphQLRequest:
def __init__(
self,
request: Union[DocumentNode, "GraphQLRequest", str],
*,
variable_values: Optional[Dict[str, Any]] = None,
operation_name: Optional[str] = None
): ...Comprehensive transport implementations for HTTP, WebSocket, and local schema operations. Includes synchronous and asynchronous variants with protocol-specific optimizations and authentication support.
# HTTP Transports
class RequestsHTTPTransport(Transport): ...
class HTTPXTransport(Transport): ...
class HTTPXAsyncTransport(AsyncTransport): ...
class AIOHTTPTransport(AsyncTransport): ...
# WebSocket Transports
class WebsocketsTransport(AsyncTransport): ...
class AIOHTTPWebsocketsTransport(AsyncTransport): ...
class PhoenixChannelWebsocketsTransport(AsyncTransport): ...
class AppSyncWebsocketsTransport(AsyncTransport): ...
# Local Schema Transport
class LocalSchemaTransport(AsyncTransport): ...
# File Upload Support
class FileVar:
def __init__(
self,
f: Any,
*,
filename: Optional[str] = None,
content_type: Optional[str] = None,
streaming: bool = False,
streaming_block_size: int = 64 * 1024
): ...Domain-specific language for programmatic GraphQL query construction without string templates. Provides type-safe query building with schema introspection, variable handling, and fragment support.
class DSLSchema:
def __init__(self, schema: GraphQLSchema): ...
class DSLQuery:
def __init__(self, *fields, **fields_with_alias): ...
class DSLMutation:
def __init__(self, *fields, **fields_with_alias): ...
class DSLSubscription:
def __init__(self, *fields, **fields_with_alias): ...
class DSLFragment:
def __init__(self, name: str): ...
def on(self, type_condition: DSLType) -> DSLFragment: ...
def select(self, *fields, **fields_with_alias) -> DSLFragment: ...
def dsl_gql(*operations, **operations_with_name) -> GraphQLRequest: ...Utilities for schema introspection, custom scalar/enum handling, variable serialization, and result parsing. Includes schema building from introspection and result processing with custom type deserialization.
def build_client_schema(introspection: IntrospectionQuery) -> GraphQLSchema: ...
def get_introspection_query_ast(**options) -> DocumentNode: ...
def parse_result(schema, document, result, operation_name=None) -> Optional[Dict[str, Any]]: ...
def serialize_variable_values(schema, document, variable_values, operation_name=None) -> Dict[str, Any]: ...
def update_schema_scalar(schema, name, scalar) -> None: ...
def update_schema_enum(schema, name, values, use_enum_values=False) -> None: ...Full-featured CLI for executing GraphQL queries, downloading schemas, and interacting with GraphQL servers. Supports multiple transport protocols, authentication methods, and output formatting.
def gql_cli() -> None: ...
def get_parser(with_examples: bool = False) -> ArgumentParser: ...
def get_transport(args: Namespace) -> Optional[AsyncTransport]: ...
def autodetect_transport(url: URL) -> str: ...Available as gql-cli command with comprehensive options for server interaction, authentication, and schema management.
Core types used across the gql library:
from typing import Union, Optional, Dict, List, Any, AsyncGenerator
from graphql import GraphQLSchema, ExecutionResult, DocumentNode, IntrospectionQuery
# Core request type
class GraphQLRequest:
document: DocumentNode
variable_values: Optional[Dict[str, Any]]
operation_name: Optional[str]
@property
def payload(self) -> Dict[str, Any]: ...
# Transport base classes
class Transport:
def execute(self, request: GraphQLRequest, *args, **kwargs) -> ExecutionResult: ...
def execute_batch(self, reqs: List[GraphQLRequest], *args, **kwargs) -> List[ExecutionResult]: ...
class AsyncTransport:
async def execute(self, request: GraphQLRequest) -> ExecutionResult: ...
async def execute_batch(self, reqs: List[GraphQLRequest], *args, **kwargs) -> List[ExecutionResult]: ...
def subscribe(self, request: GraphQLRequest) -> AsyncGenerator[ExecutionResult, None]: ...
# File upload type
class FileVar:
f: Any
filename: Optional[str]
content_type: Optional[str]
streaming: bool
streaming_block_size: int