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
—
Utilities for schema introspection, custom scalar/enum handling, variable serialization, and result parsing. Includes comprehensive tools for schema building from introspection and result processing with custom type deserialization.
Functions for building GraphQL schemas from introspection queries and generating configurable introspection queries.
def build_client_schema(introspection: IntrospectionQuery) -> GraphQLSchema:
"""
Build GraphQL schema from introspection with default directives.
Args:
introspection: IntrospectionQuery result dictionary from server
Returns:
GraphQLSchema object with proper directive definitions
Enhancement:
Adds missing @include and @skip directives to fix issue #278
where these standard directives were not included in introspection
Example:
schema = build_client_schema(introspection_result)
"""
def get_introspection_query_ast(
descriptions: bool = True,
specified_by_url: bool = False,
directive_is_repeatable: bool = False,
schema_description: bool = False,
input_value_deprecation: bool = True,
type_recursion_level: int = 7
) -> DocumentNode:
"""
Generate introspection query using DSL with configurable options.
Args:
descriptions: Include field and type descriptions
specified_by_url: Include specifiedByURL for custom scalars
directive_is_repeatable: Include directive repeatability info
schema_description: Include schema description
input_value_deprecation: Include deprecated input field info
type_recursion_level: Recursion depth for type references
Returns:
DocumentNode for introspection query
Advantage:
More configurable than graphql-core's get_introspection_query
Allows fine-tuning of introspection data retrieval
Example:
query_ast = get_introspection_query_ast(
descriptions=True,
specified_by_url=True
)
"""Functions for parsing GraphQL results with custom scalar and enum deserialization.
def parse_result(
schema: GraphQLSchema,
document: DocumentNode,
result: Optional[Dict[str, Any]],
operation_name: Optional[str] = None
) -> Optional[Dict[str, Any]]:
"""
Parse GraphQL result using schema to deserialize custom scalars/enums.
Args:
schema: GraphQL schema for type information
document: Query document for operation structure
result: Raw result from GraphQL server
operation_name: Operation name for multi-operation documents
Returns:
Parsed result with custom types properly deserialized
Features:
- Deserializes custom scalar types using their parse_value method
- Converts enum values to Python enum instances
- Handles nested objects and lists recursively
- Preserves null values and error information
Example:
parsed = parse_result(schema, query, raw_result)
# DateTime scalars are now datetime objects
# Enum values are now enum instances
"""Functions for serializing GraphQL variables using schema type information.
def serialize_variable_values(
schema: GraphQLSchema,
document: DocumentNode,
variable_values: Dict[str, Any],
operation_name: Optional[str] = None
) -> Dict[str, Any]:
"""
Serialize variable values using schema type information.
Args:
schema: GraphQL schema for type information
document: Query document with variable definitions
variable_values: Raw variable values from Python
operation_name: Operation name for multi-operation documents
Returns:
Serialized variable values ready for transport
Features:
- Serializes custom scalar types using their serialize method
- Converts Python enum instances to their values
- Handles nested input objects and lists
- Validates variable types against schema
Example:
serialized = serialize_variable_values(schema, query, {
"date": datetime.now(),
"status": StatusEnum.ACTIVE
})
# datetime is serialized to ISO string
# enum is serialized to its value
"""
def serialize_value(type_: GraphQLType, value: Any) -> Any:
"""
Serialize single value according to GraphQL type.
Args:
type_: GraphQL type definition for serialization rules
value: Python value to serialize
Returns:
Serialized value appropriate for the GraphQL type
Features:
- Handles scalar, enum, input object, and list types
- Recursive serialization for nested structures
- Custom scalar serialization using serialize method
- Enum serialization to values or names
Example:
serialized = serialize_value(
GraphQLString,
"Hello World"
)
"""Functions for updating existing GraphQL schemas with custom scalar and enum implementations.
def update_schema_scalar(
schema: GraphQLSchema,
name: str,
scalar: GraphQLScalarType
) -> None:
"""
Replace scalar type implementation in existing schema.
Args:
schema: Target GraphQL schema to modify
name: Scalar type name in schema
scalar: New scalar implementation with serialize/parse methods
Modifies:
The schema object in-place
Features:
- Updates scalar in schema's type map
- Preserves existing schema structure
- Enables custom serialization/deserialization
Example:
from graphql import GraphQLScalarType
import datetime
datetime_scalar = GraphQLScalarType(
name="DateTime",
serialize=lambda dt: dt.isoformat(),
parse_value=lambda value: datetime.fromisoformat(value)
)
update_schema_scalar(schema, "DateTime", datetime_scalar)
"""
def update_schema_scalars(
schema: GraphQLSchema,
scalars: List[GraphQLScalarType]
) -> None:
"""
Replace multiple scalar implementations in schema.
Args:
schema: Target GraphQL schema to modify
scalars: List of scalar implementations
Modifies:
The schema object in-place
Features:
- Bulk update multiple scalars efficiently
- Applies update_schema_scalar for each scalar
Example:
scalars = [datetime_scalar, uuid_scalar, decimal_scalar]
update_schema_scalars(schema, scalars)
"""
def update_schema_enum(
schema: GraphQLSchema,
name: str,
values: Union[Dict[str, Any], Type[Enum]],
use_enum_values: bool = False
) -> None:
"""
Update enum type with Python Enum or dict values.
Args:
schema: Target GraphQL schema to modify
name: Enum type name in schema
values: Python Enum class or values dictionary
use_enum_values: Whether to use enum values or enum instances
Modifies:
The schema object in-place
Features:
- Supports Python Enum classes or plain dictionaries
- Configurable value vs instance usage
- Automatic enum value mapping
Example:
from enum import Enum
class Status(Enum):
ACTIVE = "active"
INACTIVE = "inactive"
update_schema_enum(schema, "Status", Status)
# Or with dictionary
update_schema_enum(schema, "Status", {
"ACTIVE": "active",
"INACTIVE": "inactive"
})
"""Functions for debugging GraphQL AST structures and development workflows.
def node_tree(
obj: Node,
ignore_loc: bool = True,
ignore_block: bool = True,
ignored_keys: Optional[List[str]] = None
) -> str:
"""
Generate tree representation of GraphQL AST nodes for debugging.
Args:
obj: GraphQL AST node to visualize
ignore_loc: Skip location information in output
ignore_block: Skip block string attributes
ignored_keys: Additional keys to ignore in output
Returns:
String representation of node tree structure
Warning:
Output format is not guaranteed stable between versions.
Intended for development and debugging only.
Example:
from gql import gql
from gql.utilities import node_tree
query = gql('{ user { name email } }')
print(node_tree(query.document))
# Output shows AST structure:
# DocumentNode
# definitions: [
# OperationDefinitionNode
# operation: query
# selection_set:
# SelectionSetNode
# selections: [...]
# ]
"""Additional utility functions for common operations.
def to_camel_case(snake_str: str) -> str:
"""
Convert snake_case to camelCase.
Args:
snake_str: String in snake_case format
Returns:
String in camelCase format
Used by:
DSL module for automatic field name conversion
Example:
to_camel_case("first_name") # Returns "firstName"
to_camel_case("created_at") # Returns "createdAt"
"""
def str_first_element(errors: List) -> str:
"""
Extract string representation of first error from error list.
Args:
errors: List of error objects
Returns:
String representation of first error, or empty string
Features:
- Handles KeyError and TypeError when accessing first element
- Safe error message extraction
Used by:
Internal error handling utilities
Example:
errors = [GraphQLError("Field not found")]
message = str_first_element(errors) # Returns "Field not found"
"""from gql import Client
from gql.utilities import update_schema_scalar, parse_result
from graphql import GraphQLScalarType
from datetime import datetime
import json
# Define custom DateTime scalar
datetime_scalar = GraphQLScalarType(
name="DateTime",
description="DateTime scalar that serializes to ISO 8601 strings",
serialize=lambda dt: dt.isoformat() if dt else None,
parse_value=lambda value: datetime.fromisoformat(value) if value else None,
parse_literal=lambda ast: datetime.fromisoformat(ast.value) if ast.value else None
)
# Setup client with schema
transport = RequestsHTTPTransport(url="https://api.example.com/graphql")
client = Client(
transport=transport,
fetch_schema_from_transport=True,
parse_results=True # Enable automatic result parsing
)
# Update schema with custom scalar
update_schema_scalar(client.schema, "DateTime", datetime_scalar)
# Now DateTime fields are automatically parsed as datetime objects
query = gql('''
query {
posts {
title
createdAt # This will be parsed as datetime object
updatedAt # This too
}
}
''')
result = client.execute(query)
for post in result["posts"]:
created = post["createdAt"] # This is now a datetime object
print(f"Post '{post['title']}' created on {created.strftime('%Y-%m-%d')}")from gql.utilities import update_schema_enum
from enum import Enum
# Define Python enum
class PostStatus(Enum):
DRAFT = "draft"
PUBLISHED = "published"
ARCHIVED = "archived"
# Update schema with enum
update_schema_enum(client.schema, "PostStatus", PostStatus)
# Now enum fields are parsed as enum instances
query = gql('''
query {
posts {
title
status # This will be parsed as PostStatus enum
}
}
''')
result = client.execute(query)
for post in result["posts"]:
status = post["status"] # This is now a PostStatus enum instance
if status == PostStatus.PUBLISHED:
print(f"Published post: {post['title']}")
elif status == PostStatus.DRAFT:
print(f"Draft post: {post['title']}")from gql.utilities import serialize_variable_values
from datetime import datetime
from enum import Enum
class Priority(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
# Update schema with custom types
update_schema_scalar(client.schema, "DateTime", datetime_scalar)
update_schema_enum(client.schema, "Priority", Priority)
# Create mutation with custom variable types
mutation = gql('''
mutation CreateTask($input: TaskInput!) {
createTask(input: $input) {
id
title
dueDate
priority
}
}
''')
# Variables with custom types
variables = {
"input": {
"title": "Complete project",
"dueDate": datetime(2024, 12, 31, 23, 59, 59), # datetime object
"priority": Priority.HIGH # enum instance
}
}
# Serialize variables (happens automatically in client.execute)
serialized = serialize_variable_values(
client.schema,
mutation.document,
variables
)
# serialized["input"]["dueDate"] is now "2024-12-31T23:59:59"
# serialized["input"]["priority"] is now "high"
result = client.execute(mutation, variable_values=variables)from gql.utilities import get_introspection_query_ast, build_client_schema
# Generate custom introspection query
introspection_query = get_introspection_query_ast(
descriptions=True, # Include type/field descriptions
specified_by_url=True, # Include scalar specifications
directive_is_repeatable=True # Include directive info
)
# Execute introspection
introspection_request = GraphQLRequest(introspection_query)
introspection_result = client.execute(introspection_request)
# Build schema from introspection
schema = build_client_schema(introspection_result)
# Save schema for later use
with open("schema.json", "w") as f:
json.dump(introspection_result, f, indent=2)
# Load and use saved schema
with open("schema.json", "r") as f:
saved_introspection = json.load(f)
loaded_schema = build_client_schema(saved_introspection)from gql.utilities import parse_result, update_schema_scalar, update_schema_enum
from decimal import Decimal
import uuid
from datetime import datetime
# Define multiple custom scalars
decimal_scalar = GraphQLScalarType(
name="Decimal",
serialize=lambda d: str(d),
parse_value=lambda value: Decimal(str(value))
)
uuid_scalar = GraphQLScalarType(
name="UUID",
serialize=lambda u: str(u),
parse_value=lambda value: uuid.UUID(value)
)
# Update schema with all custom types
update_schema_scalar(client.schema, "DateTime", datetime_scalar)
update_schema_scalar(client.schema, "Decimal", decimal_scalar)
update_schema_scalar(client.schema, "UUID", uuid_scalar)
# Query with multiple custom scalar types
query = gql('''
query {
products {
id # UUID
name
price # Decimal
createdAt # DateTime
}
}
''')
# Execute query with automatic parsing
result = client.execute(query)
for product in result["products"]:
product_id = product["id"] # uuid.UUID object
price = product["price"] # Decimal object
created = product["createdAt"] # datetime object
print(f"Product {product_id}")
print(f" Price: ${price:.2f}")
print(f" Created: {created.strftime('%Y-%m-%d %H:%M:%S')}")from gql.utilities import parse_result
# Execute query without automatic parsing
client_no_parsing = Client(
transport=transport,
fetch_schema_from_transport=True,
parse_results=False # Disable automatic parsing
)
# Raw result has string values
raw_result = client_no_parsing.execute(query)
print(type(raw_result["products"][0]["createdAt"])) # <class 'str'>
# Manually parse result
parsed_result = parse_result(
client_no_parsing.schema,
query.document,
raw_result
)
print(type(parsed_result["products"][0]["createdAt"])) # <class 'datetime.datetime'>from gql.utilities import node_tree
from gql import gql
# Create complex query
query = gql('''
query GetUserPosts($userId: ID!, $limit: Int = 10) {
user(id: $userId) {
name
email
posts(limit: $limit) {
title
content
tags
publishedAt
}
}
}
''')
# Debug AST structure
print("Query AST Structure:")
print(node_tree(query.document))
# Output shows detailed AST tree:
# DocumentNode
# definitions: [
# OperationDefinitionNode
# operation: query
# name: NameNode(value='GetUserPosts')
# variable_definitions: [
# VariableDefinitionNode...
# ]
# selection_set: SelectionSetNode...
# ]from gql.utilities import update_schema_scalars
# Define multiple custom scalars
custom_scalars = [
GraphQLScalarType(
name="DateTime",
serialize=lambda dt: dt.isoformat(),
parse_value=lambda v: datetime.fromisoformat(v)
),
GraphQLScalarType(
name="Decimal",
serialize=lambda d: str(d),
parse_value=lambda v: Decimal(str(v))
),
GraphQLScalarType(
name="UUID",
serialize=lambda u: str(u),
parse_value=lambda v: uuid.UUID(v)
)
]
# Update all scalars at once
update_schema_scalars(client.schema, custom_scalars)
# All custom scalars are now available for parsing
query = gql('''
query {
orders {
id # UUID
total # Decimal
createdAt # DateTime
}
}
''')
result = client.execute(query)
# All custom types are automatically parsedInstall with Tessl CLI
npx tessl i tessl/pypi-gql