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

DSL Query Builder

Name: tessl/pypi-gql
Author: tessl

Domain-specific language for programmatic GraphQL query construction without string templates. Provides type-safe query building with schema introspection, variable handling, fragment support, and automatic field name conversion.

Capabilities

Core DSL Functions

Primary functions for creating and converting DSL operations into executable GraphQL requests.

def dsl_gql(*operations, **operations_with_name) -> GraphQLRequest:
    """
    Convert DSL operations into executable GraphQL requests.

    Args:
        *operations: DSLExecutable instances (DSLQuery, DSLMutation, DSLSubscription, DSLFragment)
        **operations_with_name: Same as above but with operation names as keys

    Returns:
        GraphQLRequest that can be executed by gql clients

    Example:
        dsl_gql(query) or dsl_gql(GetUser=query, CreateUser=mutation)
    """

Schema Integration

Classes for integrating with GraphQL schemas to provide type-safe query building.

class DSLSchema:
    def __init__(self, schema: GraphQLSchema):
        """
        Root class for DSL operations providing access to schema types.

        Args:
            schema: GraphQL schema object for type introspection

        Usage:
            ds = DSLSchema(client.schema)
            ds.Query.user  # Returns DSLType for User queries
            ds.User.name   # Returns DSLField for User.name field
        """
    
    # Dynamic attributes: Any schema type name becomes a DSLType instance

class DSLType:
    """
    Represents GraphQL object/interface types in DSL.
    
    Accessed via DSLSchema attributes (ds.Query, ds.User, etc.)
    Dynamic attributes: Field names become DSLField instances
    Supports snake_case to camelCase conversion for field names
    """
    
    # Dynamic attributes: Field names become DSLField instances

class DSLField:
    """
    Represents GraphQL fields with arguments and sub-selections.
    
    Accessed via DSLType attributes or method calls.
    """
    
    def args(self, **kwargs) -> DSLField:
        """
        Set field arguments.

        Args:
            **kwargs: Field arguments as keyword arguments

        Returns:
            DSLField with arguments set

        Example:
            ds.Query.user.args(id="123")
        """

    def select(self, *fields, **fields_with_alias) -> DSLField:
        """
        Select sub-fields for this field.

        Args:
            *fields: DSLField instances to select
            **fields_with_alias: DSLField instances with custom aliases

        Returns:
            DSLField with sub-field selections

        Example:
            ds.Query.user.select(ds.User.name, ds.User.email)
            ds.Query.user.select(user_name=ds.User.name)
        """

    def alias(self, alias: str) -> DSLField:
        """
        Set field alias.

        Args:
            alias: Custom alias for this field

        Returns:
            DSLField with alias set

        Example:
            ds.User.name.alias("userName")
        """

    def __call__(self, **kwargs) -> DSLField:
        """
        Shorthand for args() method.

        Args:
            **kwargs: Field arguments

        Returns:
            DSLField with arguments set

        Example:
            ds.Query.user(id="123")  # Same as ds.Query.user.args(id="123")
        """

class DSLMetaField:
    def __init__(self, name: str):
        """
        Represents GraphQL meta-fields.

        Args:
            name: Meta-field name ("__typename", "__schema", "__type")

        Example:
            DSLMetaField("__typename")
        """

Operation Classes

Classes for building different types of GraphQL operations.

class DSLQuery:
    def __init__(self, *fields, **fields_with_alias):
        """
        Represents GraphQL Query operations.

        Args:
            *fields: DSLField instances to include in query
            **fields_with_alias: DSLField instances with custom aliases

        Example:
            DSLQuery(ds.Query.user.select(ds.User.name, ds.User.email))
        """

class DSLMutation:
    def __init__(self, *fields, **fields_with_alias):
        """
        Represents GraphQL Mutation operations.

        Args:
            *fields: DSLField instances to include in mutation
            **fields_with_alias: DSLField instances with custom aliases

        Example:
            DSLMutation(ds.Mutation.createUser.args(input=user_input).select(ds.User.id))
        """

class DSLSubscription:
    def __init__(self, *fields, **fields_with_alias):
        """
        Represents GraphQL Subscription operations.

        Args:
            *fields: DSLField instances to include in subscription
            **fields_with_alias: DSLField instances with custom aliases

        Example:
            DSLSubscription(ds.Subscription.messageAdded.select(ds.Message.content))
        """

Fragment Support

Classes for creating reusable GraphQL fragments.

class DSLFragment:
    def __init__(self, name: str):
        """
        Represents named GraphQL fragments.

        Args:
            name: Fragment name

        Example:
            user_info = DSLFragment("UserInfo")
        """

    def on(self, type_condition: DSLType) -> DSLFragment:
        """
        Set fragment type condition.

        Args:
            type_condition: DSLType representing the fragment type

        Returns:
            DSLFragment with type condition set

        Example:
            user_info.on(ds.User)
        """

    def select(self, *fields, **fields_with_alias) -> DSLFragment:
        """
        Select fields for this fragment.

        Args:
            *fields: DSLField instances to include
            **fields_with_alias: DSLField instances with aliases

        Returns:
            DSLFragment with field selections

        Example:
            user_info.select(ds.User.name, ds.User.email)
        """

class DSLInlineFragment:
    def __init__(self, *fields, **fields_with_alias):
        """
        Represents inline GraphQL fragments.

        Args:
            *fields: DSLField instances to include
            **fields_with_alias: DSLField instances with aliases

        Example:
            DSLInlineFragment(ds.User.name).on(ds.User)
        """

    def on(self, type_condition: DSLType) -> DSLInlineFragment:
        """
        Set type condition for inline fragment.

        Args:
            type_condition: DSLType for fragment condition

        Returns:
            DSLInlineFragment with type condition

        Example:
            DSLInlineFragment(ds.User.name).on(ds.User)
        """

    def select(self, *fields, **fields_with_alias) -> DSLInlineFragment:
        """
        Select additional fields for inline fragment.

        Args:
            *fields: DSLField instances to add
            **fields_with_alias: DSLField instances with aliases

        Returns:
            DSLInlineFragment with additional selections
        """

Variable System

Classes for handling GraphQL variables in DSL operations.

class DSLVariableDefinitions:
    def __init__(self):
        """
        Container for operation variables.

        Dynamic attributes: Variable names become DSLVariable instances

        Example:
            vars = DSLVariableDefinitions()
            vars.user_id  # Returns DSLVariable("user_id")
        """
    
    # Dynamic attributes: Variable names become DSLVariable instances

class DSLVariable:
    """
    Represents GraphQL variables in operations.
    
    Accessed via DSLVariableDefinitions attributes.
    """
    
    def set_type(self, type_: GraphQLInputType) -> DSLVariable:
        """
        Set variable type.

        Args:
            type_: GraphQL input type for this variable

        Returns:
            DSLVariable with type set

        Example:
            vars.user_id.set_type(GraphQLNonNull(GraphQLID))
        """

    def default(self, default_value: Any) -> DSLVariable:
        """
        Set default value for variable.

        Args:
            default_value: Default value if not provided

        Returns:
            DSLVariable with default value set

        Example:
            vars.limit.default(10)
        """

Usage Examples

Basic DSL Query Building

from gql import Client
from gql.dsl import DSLSchema, DSLQuery, dsl_gql
from gql.transport.requests import RequestsHTTPTransport

# Setup client and DSL schema
transport = RequestsHTTPTransport(url="https://api.example.com/graphql")
client = Client(transport=transport, fetch_schema_from_transport=True)

# Create DSL schema from client schema
ds = DSLSchema(client.schema)

# Build query using DSL
query = DSLQuery(
    ds.Query.user(id="123").select(
        ds.User.name,
        ds.User.email,
        ds.User.posts.select(
            ds.Post.title,
            ds.Post.content,
            ds.Post.publishedAt
        )
    )
)

# Convert to executable request and execute
request = dsl_gql(query)
result = client.execute(request)

print(result["user"]["name"])

Field Arguments and Aliases

from gql.dsl import DSLSchema, DSLQuery, dsl_gql

# Build query with arguments and aliases
query = DSLQuery(
    # Field with arguments
    ds.Query.users(
        limit=10,
        offset=0,
        orderBy="CREATED_AT"
    ).select(
        ds.User.id,
        ds.User.name,
        # Field with alias
        user_email=ds.User.email,
        # Nested field with arguments
        ds.User.posts(published=True).select(
            ds.Post.title,
            published_date=ds.Post.publishedAt
        )
    )
)

request = dsl_gql(query)
result = client.execute(request)

# Access aliased fields
for user in result["users"]:
    print(f"{user['name']} - {user['user_email']}")

Variables and Dynamic Queries

from gql.dsl import DSLSchema, DSLQuery, DSLVariableDefinitions, dsl_gql

# Setup variables
vars = DSLVariableDefinitions()

# Build query with variables
query = DSLQuery(
    ds.Query.user(id=vars.user_id).select(
        ds.User.name,
        ds.User.email,
        ds.User.posts(limit=vars.post_limit).select(
            ds.Post.title,
            ds.Post.content
        )
    )
)

# Execute with different variable values
request = dsl_gql(query)

result1 = client.execute(request, variable_values={
    "user_id": "123",
    "post_limit": 5
})

result2 = client.execute(request, variable_values={
    "user_id": "456", 
    "post_limit": 10
})

Fragment Usage

from gql.dsl import DSLSchema, DSLQuery, DSLFragment, dsl_gql

# Define reusable fragment
user_info = DSLFragment("UserInfo").on(ds.User).select(
    ds.User.id,
    ds.User.name,
    ds.User.email,
    ds.User.createdAt
)

# Use fragment in multiple places
query = DSLQuery(
    # Use fragment in different contexts
    current_user=ds.Query.currentUser.select(user_info),
    user_by_id=ds.Query.user(id="123").select(user_info),
    
    # Use fragment in nested selections
    ds.Query.posts.select(
        ds.Post.title,
        ds.Post.author.select(user_info)
    )
)

# Include fragment in request
request = dsl_gql(query, user_info)
result = client.execute(request)

Inline Fragments for Union Types

from gql.dsl import DSLSchema, DSLQuery, DSLInlineFragment, dsl_gql

# Handle union types with inline fragments
query = DSLQuery(
    ds.Query.search(query="python").select(
        # Common fields available on all union members
        ds.SearchResult.id,
        
        # Type-specific fields using inline fragments
        DSLInlineFragment().on(ds.User).select(
            ds.User.name,
            ds.User.email
        ),
        
        DSLInlineFragment().on(ds.Post).select(
            ds.Post.title,
            ds.Post.content
        ),
        
        DSLInlineFragment().on(ds.Repository).select(
            ds.Repository.name,
            ds.Repository.description,
            ds.Repository.starCount
        )
    )
)

request = dsl_gql(query)
result = client.execute(request)

# Results will include type-specific fields based on actual types
for item in result["search"]:
    if "name" in item and "email" in item:
        print(f"User: {item['name']} ({item['email']})")
    elif "title" in item:
        print(f"Post: {item['title']}")
    elif "starCount" in item:
        print(f"Repository: {item['name']} ({item['starCount']} stars)")

Mutations with Input Objects

from gql.dsl import DSLSchema, DSLMutation, dsl_gql

# Build mutation with input object
user_input = {
    "name": "John Doe",
    "email": "john@example.com",
    "age": 30
}

mutation = DSLMutation(
    ds.Mutation.createUser(input=user_input).select(
        ds.User.id,
        ds.User.name,
        ds.User.email,
        ds.User.createdAt
    )
)

request = dsl_gql(mutation)
result = client.execute(request)

created_user = result["createUser"]
print(f"Created user {created_user['name']} with ID {created_user['id']}")

Subscriptions with DSL

import asyncio
from gql.dsl import DSLSchema, DSLSubscription, dsl_gql

async def handle_subscription():
    # Build subscription
    subscription = DSLSubscription(
        ds.Subscription.messageAdded(channelId="general").select(
            ds.Message.id,
            ds.Message.content,
            ds.Message.user.select(
                ds.User.name,
                ds.User.avatar
            ),
            ds.Message.timestamp
        )
    )

    request = dsl_gql(subscription)
    
    async with client.connect_async() as session:
        async for result in session.subscribe(request):
            message = result["messageAdded"]
            user = message["user"]
            print(f"[{message['timestamp']}] {user['name']}: {message['content']}")

asyncio.run(handle_subscription())

Meta-fields and Introspection

from gql.dsl import DSLSchema, DSLQuery, DSLMetaField, dsl_gql

# Query with typename meta-field
query = DSLQuery(
    ds.Query.search(query="graphql").select(
        DSLMetaField("__typename"),  # Include __typename for each result
        ds.SearchResult.id,
        
        # Conditional selection based on type
        DSLInlineFragment().on(ds.User).select(
            ds.User.name
        ),
        
        DSLInlineFragment().on(ds.Post).select(
            ds.Post.title
        )
    )
)

request = dsl_gql(query)
result = client.execute(request)

# Use __typename to handle different types
for item in result["search"]:
    if item["__typename"] == "User":
        print(f"User: {item['name']}")
    elif item["__typename"] == "Post":
        print(f"Post: {item['title']}")

Complex Nested Queries

from gql.dsl import DSLSchema, DSLQuery, DSLFragment, dsl_gql

# Define fragments for reusability
post_summary = DSLFragment("PostSummary").on(ds.Post).select(
    ds.Post.id,
    ds.Post.title,
    ds.Post.publishedAt,
    ds.Post.commentCount,
    ds.Post.likeCount
)

user_profile = DSLFragment("UserProfile").on(ds.User).select(
    ds.User.id,
    ds.User.name,
    ds.User.email,
    ds.User.bio,
    ds.User.followersCount,
    ds.User.followingCount
)

# Build complex nested query
query = DSLQuery(
    # Current user with profile info
    ds.Query.currentUser.select(
        user_profile,
        
        # Recent posts with summary
        ds.User.posts(limit=5, orderBy="PUBLISHED_AT_DESC").select(
            post_summary,
            
            # Comments on each post
            ds.Post.comments(limit=3).select(
                ds.Comment.id,
                ds.Comment.content,
                ds.Comment.author.select(user_profile)
            )
        ),
        
        # Following users
        ds.User.following(limit=10).select(user_profile)
    )
)

request = dsl_gql(query, post_summary, user_profile)
result = client.execute(request)

# Access nested data
current_user = result["currentUser"]
print(f"User: {current_user['name']}")
print(f"Recent posts: {len(current_user['posts'])}")

for post in current_user["posts"]:
    print(f"  - {post['title']} ({post['commentCount']} comments)")
    for comment in post["comments"]:
        print(f"    * {comment['author']['name']}: {comment['content']}")

Snake Case to Camel Case Conversion

from gql.dsl import DSLSchema, DSLQuery, dsl_gql

# DSL automatically converts snake_case to camelCase
query = DSLQuery(
    ds.Query.current_user.select(  # Becomes currentUser
        ds.User.first_name,        # Becomes firstName
        ds.User.last_name,         # Becomes lastName
        ds.User.created_at,        # Becomes createdAt
        ds.User.updated_at         # Becomes updatedAt
    )
)

# Generated GraphQL uses camelCase field names
request = dsl_gql(query)
print(request.payload["query"])
# Output includes: currentUser { firstName lastName createdAt updatedAt }

result = client.execute(request)
user = result["currentUser"]
print(f"{user['firstName']} {user['lastName']}")

Install with Tessl CLI