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
—
Full-featured CLI for executing GraphQL queries, downloading schemas, and interacting with GraphQL servers. Supports multiple transport protocols, authentication methods, and comprehensive output formatting.
Core functions for building and running the command-line interface.
def gql_cli() -> None:
"""
Main CLI entry point registered as console script.
Available as the 'gql-cli' command after package installation.
Provides complete GraphQL client functionality from command line.
Features:
- Execute GraphQL queries, mutations, and subscriptions
- Download and print GraphQL schemas via introspection
- Multiple transport protocol support
- Authentication for AWS AppSync
- Configurable logging and debugging
- JSON variable input support
Usage:
gql-cli https://api.example.com/graphql
gql-cli wss://api.example.com/graphql -s
"""
def get_parser(with_examples: bool = False) -> ArgumentParser:
"""
Create argument parser for CLI.
Args:
with_examples: Include usage examples in help text
Returns:
Configured ArgumentParser instance with all CLI options
Used by:
gql_cli() function and testing infrastructure
"""
def get_transport(args: Namespace) -> Optional[AsyncTransport]:
"""
Create transport instance from CLI arguments.
Args:
args: Parsed command line arguments
Returns:
AsyncTransport instance based on URL scheme and transport options
Features:
- Auto-detection of transport type from URL
- Support for HTTP, WebSocket, and specialized transports
- Authentication configuration for AppSync
- Custom transport selection override
Supported Transports:
- aiohttp: HTTP transport using aiohttp
- httpx: HTTP transport using httpx
- websockets: WebSocket transport for subscriptions
- phoenix: Phoenix Channel WebSocket transport
- appsync: AWS AppSync WebSocket transport
"""
def autodetect_transport(url: URL) -> str:
"""
Determine appropriate transport based on URL scheme.
Args:
url: URL object with scheme information
Returns:
Transport name string
Logic:
- ws/wss schemes → "websockets"
- http/https schemes → "aiohttp" (default HTTP transport)
Example:
autodetect_transport(URL("wss://api.example.com/graphql")) # Returns "websockets"
autodetect_transport(URL("https://api.example.com/graphql")) # Returns "aiohttp"
"""
def get_transport_args(args: Namespace) -> Dict[str, Any]:
"""
Extract transport configuration from CLI arguments.
Args:
args: Parsed command line arguments
Returns:
Dictionary of transport configuration options
Extracts:
- HTTP headers from --headers arguments
- Transport-specific configuration options
"""
def get_execute_args(args: Namespace) -> Dict[str, Any]:
"""
Extract execution arguments from CLI arguments.
Args:
args: Parsed command line arguments
Returns:
Dictionary of execution options
Extracts:
- operation_name: Specific operation name for multi-operation documents
- variable_values: Parsed JSON variables from --variables arguments
"""
def get_introspection_args(args: Namespace) -> Dict[str, Any]:
"""
Parse schema download options from CLI arguments.
Args:
args: Parsed command line arguments
Returns:
Dictionary of introspection query configuration
Options:
- descriptions: Include field/type descriptions
- specified_by_url: Include specifiedByURL for scalars
- directive_is_repeatable: Include directive repeatability info
- schema_description: Include schema description
- input_value_deprecation: Include deprecated input fields
- type_recursion_level: Recursion depth for type references
"""
def positive_int_or_none(value_str: str) -> Optional[int]:
"""
Parse CLI integer arguments that can be None.
Args:
value_str: String value from command line
Returns:
Positive integer or None
Accepts:
- Positive integers: "10", "100"
- None string: "none", "None"
Raises:
argparse.ArgumentTypeError: If value is invalid
Used by:
Timeout and limit argument parsing
"""server # GraphQL endpoint URL
# Examples: https://api.example.com/graphql
# wss://api.example.com/graphql--variables, -V # Query variables as key:json_value pairs
# Format: key:json_value
# Examples: -V id:123 -V name:'"John"' -V active:true
--operation-name, -o # Operation name for multi-operation documents
# Used when GraphQL document contains multiple operations
--execute-timeout # Client execution timeout in seconds
# Default: 10 seconds--print-schema # Download and print schema via introspection
# Outputs complete GraphQL schema definition
--schema-download # Configure introspection query options
# Comma-separated list of options:
# - descriptions (default: true)
# - specified_by_url (default: false)
# - directive_is_repeatable (default: false)
# - schema_description (default: false)
# - input_value_deprecation (default: true)
# - type_recursion_level:N (default: 7)--transport # Choose specific transport
# Options: auto (default), aiohttp, httpx, websockets, phoenix, appsync
# Auto-detection based on URL scheme if not specified
--headers, -H # HTTP headers as key:value pairs
# Format: key:value
# Examples: -H "Authorization:Bearer token123"
# -H "Content-Type:application/json"--api-key # AppSync API key authentication
# Used with AppSync GraphQL endpoints
--jwt # AppSync JWT authentication
# JWT token for Cognito User Pools or OIDC--debug, -d # Enable debug logging
# Shows detailed transport and execution information
--verbose, -v # Enable info logging
# Shows general operation information# Execute simple query
gql-cli https://api.example.com/graphql <<< 'query { hello }'
# Execute query with variables
gql-cli https://api.example.com/graphql \
-V id:123 -V name:'"John Doe"' \
<<< 'query GetUser($id: ID!, $name: String) { user(id: $id, name: $name) { id name email } }'
# Execute specific operation from multi-operation document
gql-cli https://api.example.com/graphql \
--operation-name GetUser \
<<< 'query GetUser { user { name } } query GetPosts { posts { title } }'# Download and print complete schema
gql-cli https://api.example.com/graphql --print-schema
# Download schema with custom introspection options
gql-cli https://api.example.com/graphql --print-schema \
--schema-download descriptions,specified_by_url,directive_is_repeatable
# Save schema to file
gql-cli https://api.example.com/graphql --print-schema > schema.graphql# Basic authentication with headers
gql-cli https://api.example.com/graphql \
-H "Authorization:Bearer your-jwt-token" \
-H "User-Agent:MyApp/1.0" \
<<< 'query { currentUser { name } }'
# Force specific HTTP transport
gql-cli https://api.example.com/graphql \
--transport httpx \
-H "API-Key:your-api-key" \
<<< 'query { data }'
# With timeout configuration
gql-cli https://api.example.com/graphql \
--execute-timeout 30 \
<<< 'query { slowOperation }'# Basic WebSocket subscription
gql-cli wss://api.example.com/graphql \
<<< 'subscription { messageAdded { id content user } }'
# WebSocket with headers for authentication
gql-cli wss://api.example.com/graphql \
-H "Authorization:Bearer your-token" \
<<< 'subscription { notifications { type message } }'
# Force WebSocket transport (useful for dual HTTP/WS endpoints)
gql-cli https://api.example.com/graphql \
--transport websockets \
<<< 'subscription { events { id data } }'# Phoenix Framework Absinthe GraphQL server
gql-cli wss://phoenix-app.example.com/socket/websocket \
--transport phoenix \
<<< 'subscription { chatMessage { content user } }'
# Phoenix with custom channel (if supported by server)
gql-cli wss://phoenix-app.example.com/socket/websocket \
--transport phoenix \
-V channel:'"custom_channel"' \
<<< 'subscription { updates }'# AppSync with API key authentication
gql-cli https://your-api.appsync-api.us-east-1.amazonaws.com/graphql \
--transport appsync \
--api-key your-api-key \
<<< 'subscription { onCommentAdded { id content } }'
# AppSync with JWT authentication (Cognito User Pools)
gql-cli https://your-api.appsync-api.us-east-1.amazonaws.com/graphql \
--transport appsync \
--jwt your-cognito-jwt-token \
<<< 'subscription { onPostUpdated { id title } }'
# AppSync with IAM authentication (uses AWS credentials from environment)
gql-cli https://your-api.appsync-api.us-east-1.amazonaws.com/graphql \
--transport appsync \
<<< 'subscription { onDataChange { id value } }'# Variables with different JSON types
gql-cli https://api.example.com/graphql \
-V userId:123 \
-V isActive:true \
-V tags:'["python", "graphql"]' \
-V metadata:'{"source": "cli", "version": 1}' \
<<< 'query GetUser($userId: ID!, $isActive: Boolean!, $tags: [String!]!, $metadata: JSON) {
user(id: $userId, active: $isActive) {
name
tags(filter: $tags)
metadata(input: $metadata)
}
}'
# Complex nested input objects
gql-cli https://api.example.com/graphql \
-V input:'{"name": "John Doe", "email": "john@example.com", "profile": {"age": 30, "city": "NYC"}}' \
<<< 'mutation CreateUser($input: UserInput!) {
createUser(input: $input) {
id
name
email
}
}'# Enable debug logging to see transport details
gql-cli https://api.example.com/graphql --debug \
<<< 'query { debug }'
# Enable verbose logging for general information
gql-cli https://api.example.com/graphql --verbose \
<<< 'query { status }'
# Combine with schema download for development
gql-cli https://api.example.com/graphql --debug --print-schema \
--schema-download descriptions,specified_by_url > debug_schema.graphql# Read query from file
gql-cli https://api.example.com/graphql < query.graphql
# Save results to file
gql-cli https://api.example.com/graphql \
<<< 'query { users { id name } }' > results.json
# Pipeline with other tools
cat complex_query.graphql | \
gql-cli https://api.example.com/graphql \
-V limit:10 | \
jq '.data.users[].name'# Using heredoc for complex queries
gql-cli https://api.example.com/graphql <<EOF
query GetUserWithPosts {
user(id: "123") {
id
name
email
posts(limit: 5) {
id
title
content
publishedAt
comments(limit: 3) {
id
content
author {
name
}
}
}
}
}
EOF# Common pattern: alias for authenticated API
alias api-cli='gql-cli https://api.example.com/graphql -H "Authorization:Bearer $API_TOKEN"'
# Use the alias
api-cli <<< 'query { currentUser { name } }'
# Environment variable for dynamic configuration
export GQL_ENDPOINT="https://api.example.com/graphql"
export GQL_TOKEN="your-token-here"
gql-cli "$GQL_ENDPOINT" \
-H "Authorization:Bearer $GQL_TOKEN" \
<<< 'query { profile }'The CLI outputs errors to stderr and results to stdout, enabling proper shell scripting:
# Capture only results
gql-cli https://api.example.com/graphql <<< 'query { data }' > results.json
# Capture only errors
gql-cli https://api.example.com/graphql <<< 'query { invalid }' 2> errors.log
# Handle errors in scripts
if ! gql-cli https://api.example.com/graphql <<< 'query { health }' >/dev/null 2>&1; then
echo "API is down"
exit 1
fi#!/bin/bash
# Check API health and get user count
API_URL="https://api.example.com/graphql"
AUTH_HEADER="Authorization:Bearer $API_TOKEN"
# Check if API is healthy
if ! gql-cli "$API_URL" -H "$AUTH_HEADER" <<< 'query { health }' >/dev/null 2>&1; then
echo "API health check failed"
exit 1
fi
# Get user count
USER_COUNT=$(gql-cli "$API_URL" -H "$AUTH_HEADER" \
<<< 'query { userCount }' | jq -r '.data.userCount')
echo "Current user count: $USER_COUNT"# GitHub Actions example
- name: Test GraphQL API
run: |
gql-cli https://api.example.com/graphql \
-H "Authorization:Bearer ${{ secrets.API_TOKEN }}" \
<<< 'query { health { status version } }' | \
jq -e '.data.health.status == "ok"'#!/bin/bash
# Monitor GraphQL subscription and log messages
LOG_FILE="/var/log/graphql-messages.log"
API_URL="wss://api.example.com/graphql"
# Continuous monitoring with restart on failure
while true; do
echo "$(date): Starting GraphQL subscription monitor" >> "$LOG_FILE"
gql-cli "$API_URL" \
-H "Authorization:Bearer $MONITOR_TOKEN" \
<<< 'subscription { alerts { level message timestamp } }' | \
while IFS= read -r line; do
echo "$(date): $line" >> "$LOG_FILE"
done
echo "$(date): Subscription ended, restarting in 5s" >> "$LOG_FILE"
sleep 5
doneInstall with Tessl CLI
npx tessl i tessl/pypi-gql