CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-gql

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

Pending
Overview
Eval results
Files

utilities.mddocs/

Schema Utilities

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.

Capabilities

Schema Building and Introspection

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
        )
    """

Result Processing

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
    """

Variable Serialization

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"
        )
    """

Schema Manipulation

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"
        })
    """

Development and Debugging Utilities

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: [...]
        #   ]
    """

Utility Functions

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"
    """

Usage Examples

Custom Scalar Implementation

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')}")

Custom Enum Handling

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']}")

Variable Serialization with Custom Types

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)

Schema Introspection and Building

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)

Result Parsing with Multiple Custom Types

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')}")

Manual Result Parsing

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'>

AST Debugging

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...
#   ]

Multiple Scalar Updates

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 parsed

Install with Tessl CLI

npx tessl i tessl/pypi-gql

docs

cli.md

client-sessions.md

dsl.md

index.md

transports.md

utilities.md

tile.json