### Install GQL pre-releases Source: https://github.com/graphql-python/gql/blob/master/docs/intro.md Install the latest pre-release versions of GQL and its dependencies. ```bash pip install --pre "gql[all]" ``` -------------------------------- ### Install GQL with all dependencies using conda Source: https://github.com/graphql-python/gql/blob/master/docs/intro.md Install GQL and all its extra dependencies using conda. ```bash conda install gql-with-all ``` -------------------------------- ### Install GQL with all dependencies Source: https://github.com/graphql-python/gql/blob/master/docs/intro.md Use this command to install GQL along with all extra dependencies for various transports. ```bash pip install "gql[all]" ``` -------------------------------- ### Install GQL pre-releases with conda Source: https://github.com/graphql-python/gql/blob/master/docs/intro.md Install the latest pre-release versions of GQL and graphql-core using conda. ```bash conda install -c conda-forge -c conda-forge/label/graphql_core_alpha -c conda-forge/label/gql_beta gql-with-all ``` -------------------------------- ### Install Tox Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Install the tox testing tool for managing multiple Python environments. ```console pip install tox ``` -------------------------------- ### Install GQL with AIOHTTP transport Source: https://github.com/graphql-python/gql/blob/master/docs/intro.md Install GQL with only the aiohttp dependency for the AIOHTTPTransport. ```bash pip install gql[aiohttp] ``` -------------------------------- ### Example GraphQL Query Result Source: https://github.com/graphql-python/gql/blob/master/README.md The expected output when executing the basic sync GraphQL query example. ```bash $ python basic_example.py {'continents': [{'code': 'AF', 'name': 'Africa'}, {'code': 'AN', 'name': 'Antarctica'}, {'code': 'AS', 'name': 'Asia'}, {'code': 'EU', 'name': 'Europe'}, {'code': 'NA', 'name': 'North America'}, {'code': 'OC', 'name': 'Oceania'}, {'code': 'SA', 'name': 'South America'}]} ``` -------------------------------- ### Install Development Dependencies Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Install the project's development dependencies within your activated virtual environment. ```console python -m pip install -e.[dev] ``` -------------------------------- ### Install Tox-Conda Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Install the tox-conda plugin to enable running tox with Conda environments. ```sh conda install -c conda-forge tox-conda ``` -------------------------------- ### Async GraphQL Query with gql Source: https://github.com/graphql-python/gql/blob/master/README.md Demonstrates how to set up an asynchronous HTTP transport, create a gql client, and execute a query asynchronously. Ensure aiohttp is installed for this transport. ```python import asyncio from gql import Client, gql from gql.transport.aiohttp import AIOHTTPTransport async def main(): # Select your transport with a defined url endpoint transport = AIOHTTPTransport(url="https://countries.trevorblades.com/graphql") # Create a GraphQL client using the defined transport client = Client(transport=transport) # Provide a GraphQL query query = gql( """ query getContinents { continents { code name } } """ ) # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with client as session: # Execute the query result = await session.execute(query) print(result) asyncio.run(main()) ``` -------------------------------- ### Install Dependencies in Conda Environment Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md After activating the Conda environment, install all necessary dependencies. ```console pip install -e.[dev] ``` -------------------------------- ### Full Example with API Key Authentication Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Demonstrates setting up and using the AppSyncWebsocketsTransport with API key authentication for executing subscriptions. Ensure AWS_GRAPHQL_API_ENDPOINT and AWS_GRAPHQL_API_KEY environment variables are set. ```python import asyncio import os import sys from urllib.parse import urlparse from gql import Client, gql from gql.transport.appsync_auth import AppSyncApiKeyAuthentication from gql.transport.appsync_websockets import AppSyncWebsocketsTransport # Uncomment the following lines to enable debug output # import logging # logging.basicConfig(level=logging.DEBUG) async def main(): # Should look like: # https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql url = os.environ.get("AWS_GRAPHQL_API_ENDPOINT") api_key = os.environ.get("AWS_GRAPHQL_API_KEY") if url is None or api_key is None: print("Missing environment variables") sys.exit() # Extract host from url host = str(urlparse(url).netloc) print(f"Host: {host}") auth = AppSyncApiKeyAuthentication(host=host, api_key=api_key) transport = AppSyncWebsocketsTransport(url=url, auth=auth) async with Client(transport=transport) as session: subscription = gql( """ subscription onCreateMessage { onCreateMessage { message } } """ ) print("Waiting for messages...") async for result in session.subscribe(subscription): print(result) asyncio.run(main()) ``` -------------------------------- ### Comprehensive Directive Example Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Illustrates directives on multiple locations within a single query, including operation, field, fragment spread, inline fragment, and meta field. ```python from gql.dsl import DSLFragment, DSLInlineFragment, DSLQuery, dsl_gql # Create variables for directive conditions var = DSLVariableDefinitions() # Fragment with directive on definition character_fragment = DSLFragment("CharacterInfo").on(ds.Character).select( ds.Character.name, ds.Character.appearsIn ).directives(ds("@fragmentDefinition")) # Query with directives on multiple locations query = DSLQuery( ds.Query.hero.args(episode=var.episode).select( # Field with directive ds.Character.name.directives(ds("@skip").args(**{"if": var.skipName})), # Fragment spread with directive character_fragment.spread().directives( ds("@include").args(**{"if": var.includeFragment}) ), # Inline fragment with directive DSLInlineFragment().on(ds.Human).select(ds.Human.homePlanet).directives( ds("@skip").args(**{"if": var.skipHuman}) ), # Meta field with directive DSLMetaField("__typename").directives( ds("@include").args(**{"if": var.includeType}) ) ) ).directives(ds("@query")) # Operation directive ``` -------------------------------- ### Execute GraphQL Query with HTTPXAsyncTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/httpx_async.md Demonstrates how to set up and use the HTTPXAsyncTransport to execute a GraphQL query. Ensure the httpx library is installed. GraphQL subscriptions are not supported. ```python import asyncio from gql import Client, gql from gql.transport.httpx import HTTPXAsyncTransport async def main(): transport = HTTPXAsyncTransport(url="https://countries.trevorblades.com/graphql") # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with Client( transport=transport, fetch_schema_from_transport=True, ) as session: # Execute single query query = gql( """ query getContinents { continents { code name } } """ ) result = await session.execute(query) print(result) asyncio.run(main()) ``` -------------------------------- ### Upload File with aiohttp and gql Source: https://github.com/graphql-python/gql/blob/master/docs/usage/file_upload.md Use this snippet to upload a file by first downloading it with aiohttp and then passing the content as a FileVar to the GraphQL mutation. Ensure you have aiohttp and gql installed. ```python async with aiohttp.ClientSession() as http_client: async with http_client.get('YOUR_DOWNLOAD_URL') as resp: # We now have a StreamReader instance in resp.content # and we provide it to the variable_values attribute of the query transport = AIOHTTPTransport(url='YOUR_GRAPHQL_URL') client = Client(transport=transport) query = gql(''' mutation($file: Upload!) { singleUpload(file: $file) { id } } ''') query.variable_values = {"file": FileVar(resp.content)} result = client.execute(query, upload_files=True) ``` -------------------------------- ### Using HTTPXTransport for GraphQL Queries Source: https://github.com/graphql-python/gql/blob/master/docs/transports/httpx.md Demonstrates how to set up and use the HTTPXTransport to execute a GraphQL query. Ensure httpx is installed. The client can fetch the schema directly from the transport. ```python from gql import Client, gql from gql.transport.httpx import HTTPXTransport transport = HTTPXTransport(url="https://countries.trevorblades.com/") client = Client(transport=transport, fetch_schema_from_transport=True) query = gql( """ query getContinents { continents { code name } } """ ) result = client.execute(query) print(result) ``` -------------------------------- ### AppSync API Key Authentication Setup Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Instantiates the AppSyncApiKeyAuthentication class and the AppSyncWebsocketsTransport for API key authentication. Replace placeholders with your actual API endpoint and key. ```python auth = AppSyncApiKeyAuthentication( host="XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com", api_key="YOUR_API_KEY", ) transport = AppSyncWebsocketsTransport( url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql", auth=auth, ) ``` -------------------------------- ### AppSync IAM Authentication Setup (Default) Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Sets up the AppSyncWebsocketsTransport for IAM authentication without explicitly providing credentials. It defaults to using environment variables or AWS configuration files. ```python transport = AppSyncWebsocketsTransport( url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql", ) ``` -------------------------------- ### Async GraphQL Query Execution with DSL Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md This example shows how to execute a GraphQL query asynchronously using the gql-python client and its DSL. It includes setting up an AIOHTTP transport, creating a client, and executing a query to fetch continent data. ```python import asyncio from gql import Client from gql.dsl import DSLQuery, DSLSchema, dsl_gql from gql.transport.aiohttp import AIOHTTPTransport async def main(): transport = AIOHTTPTransport(url="https://countries.trevorblades.com/graphql") client = Client(transport=transport, fetch_schema_from_transport=True) # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection. # Because we requested to fetch the schema from the transport, # GQL will fetch the schema just after the establishment of the first session async with client as session: assert client.schema is not None # Instantiate the root of the DSL Schema as ds ds = DSLSchema(client.schema) # Create the query using dynamically generated attributes from ds query = dsl_gql( DSLQuery( ds.Query.continents(filter={"code": {"eq": "EU"}}).select( ds.Continent.code, ds.Continent.name ) ) ) result = await session.execute(query) print(result) # This can also be written as: # I want to query the continents query_continents = ds.Query.continents # I want to get only the continents with code equal to "EU" query_continents(filter={"code": {"eq": "EU"}}) # I want this query to return the code and name fields query_continents.select(ds.Continent.code) query_continents.select(ds.Continent.name) # I generate a document from my query to be able to execute it query = dsl_gql(DSLQuery(query_continents)) # Execute the query result = await session.execute(query) print(result) asyncio.run(main()) ``` -------------------------------- ### Access Root Types (Query, Mutation, Subscription) Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Access root types like Query, Mutation, or Subscription from the DSLSchema instance to start building your query. ```python ds.Query ``` -------------------------------- ### Connect and Execute Queries with AIOHTTPWebsocketsTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/aiohttp_websockets.md Use this snippet to establish a WebSocket connection using AIOHTTPWebsocketsTransport and execute a single query. The `async with Client` context manager handles connection setup and teardown. ```python import asyncio import logging from gql import Client, gql from gql.transport.aiohttp_websockets import AIOHTTPWebsocketsTransport logging.basicConfig(level=logging.INFO) async def main(): transport = AIOHTTPWebsocketsTransport( url="wss://countries.trevorblades.com/graphql" ) # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with Client( transport=transport, ) as session: # Execute single query query = gql( """ query getContinents { continents { code name } } """ ) result = await session.execute(query) print(result) # Request subscription subscription = gql( """ subscription { somethingChanged { id } } """ ) async for result in session.subscribe(subscription): print(result) asyncio.run(main()) ``` -------------------------------- ### Synchronous GraphQL Query Execution with DSL Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md This example demonstrates how to execute a GraphQL query synchronously using the gql-python client and its DSL. It sets up a RequestsHTTPTransport, creates a client, and executes a query to fetch continent data. ```python from gql import Client from gql.dsl import DSLQuery, DSLSchema, dsl_gql from gql.transport.requests import RequestsHTTPTransport transport = RequestsHTTPTransport( url="https://countries.trevorblades.com/", verify=True, retries=3, ) client = Client(transport=transport, fetch_schema_from_transport=True) # Using `with` on the sync client will start a connection on the transport # and provide a `session` variable to execute queries on this connection. # Because we requested to fetch the schema from the transport, # GQL will fetch the schema just after the establishment of the first session with client as session: # We should have received the schema now that the session is established assert client.schema is not None # Instantiate the root of the DSL Schema as ds ds = DSLSchema(client.schema) # Create the query using dynamically generated attributes from ds query = dsl_gql( DSLQuery(ds.Query.continents.select(ds.Continent.code, ds.Continent.name)) ) result = session.execute(query) print(result) ``` -------------------------------- ### AppSync IAM Authentication Setup (Manual Credentials) Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Configures IAM authentication for AppSyncWebsocketsTransport by manually providing AWS credentials and region. This is an alternative to relying on default credential discovery. ```python from botocore.credentials import Credentials credentials = Credentials( access_key = os.environ.get("AWS_ACCESS_KEY_ID"), secret_key= os.environ.get("AWS_SECRET_ACCESS_KEY"), token=os.environ.get("AWS_SESSION_TOKEN", None), # Optional ) auth = AppSyncIAMAuthentication( host="XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com", credentials=credentials, region_name="your region" ) transport = AppSyncWebsocketsTransport( url="https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql", auth=auth, ) ``` -------------------------------- ### Construct a Mutation Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Define mutations by starting from `ds.Mutation` and using `DSLMutation`. Arguments can be passed using the `args` method. ```python query = dsl_gql( DSLMutation( ds.Mutation.createReview.args( episode=6, review={\"stars\": 5, \"commentary\": \"This is a great movie!\"} ).select(ds.Review.stars, ds.Review.commentary) ) ) ``` -------------------------------- ### Define Subscription Operation Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Construct a GraphQL subscription operation starting from ds.Subscription and using DSLSubscription. ```python query = dsl_gql( DSLSubscription( ds.Subscription.reviewAdded(episode=6).select(ds.Review.stars, ds.Review.commentary) ) ) ``` -------------------------------- ### Configure Connection Retries with Tenacity Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/async_permanent_session.md Customize the connection retry behavior by providing a `tenacity.retry` decorator to the `retry_connect` argument. This example sets a maximum wait time of 5 minutes between connection attempts. ```python from tenacity import retry, retry_if_exception_type, wait_exponential # Here wait maximum 5 minutes between connection retries retry_connect = retry( # which exceptions should cause a retry (here: everything) retry=retry_if_exception_type(Exception), wait=wait_exponential(max=300), # max wait time in seconds ) session = await client.connect_async( reconnecting=True, retry_connect=retry_connect, ) ``` -------------------------------- ### AppSync API Key Authentication with HTTP Transport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Set up AppSync API Key authentication for HTTP requests using AIOHTTPTransport. This example uses environment variables for the API endpoint and key. Ensure the AWS_GRAPHQL_API_ENDPOINT and AWS_GRAPHQL_API_KEY environment variables are set. ```python import asyncio import os import sys from urllib.parse import urlparse from gql import Client, gql from gql.transport.aiohttp import AIOHTTPTransport from gql.transport.appsync_auth import AppSyncApiKeyAuthentication # Uncomment the following lines to enable debug output # import logging # logging.basicConfig(level=logging.DEBUG) async def main(): # Should look like: # https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql url = os.environ.get("AWS_GRAPHQL_API_ENDPOINT") api_key = os.environ.get("AWS_GRAPHQL_API_KEY") if url is None or api_key is None: print("Missing environment variables") sys.exit() # Extract host from url host = str(urlparse(url).netloc) auth = AppSyncApiKeyAuthentication(host=host, api_key=api_key) transport = AIOHTTPTransport(url=url, auth=auth) async with Client( transport=transport, fetch_schema_from_transport=False, ) as session: query = gql( """ mutation createMessage($message: String!) { createMessage(input: {message: $message}) { id message createdAt } }""" ) query.variable_values = {"message": "Hello world!"} result = await session.execute(query) print(result) asyncio.run(main()) ``` -------------------------------- ### Configure Execution Retries with Tenacity Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/async_permanent_session.md Control the retry behavior for execution calls by providing a `tenacity.retry` decorator to the `retry_execute` argument. This example limits execution retries to 3 attempts. ```python from tenacity import ( retry, retry_if_exception_type, stop_after_attempt, wait_exponential, ) # Here Only 3 tries for execute calls retry_execute = retry( retry=retry_if_exception_type(Exception), stop=stop_after_attempt(3), wait=wait_exponential(), ) session = await client.connect_async( reconnecting=True, retry_execute=retry_execute, ) ``` -------------------------------- ### Custom Subscription Retry Logic Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/async_permanent_session.md Implement custom retry logic for subscriptions by applying a `tenacity.retry` decorator to your subscription handling method. This example retries on most exceptions but avoids retrying `TransportQueryError`. ```python from tenacity import ( retry, retry_if_exception_type, retry_unless_exception_type, stop_after_attempt, wait_exponential, ) from gql.transport.exceptions import TransportQueryError @retry( retry=retry_if_exception_type(Exception) & retry_unless_exception_type(TransportQueryError), stop=stop_after_attempt(3), wait=wait_exponential(), ) async def execute_subscription1(session): async for result in session.subscribe(subscription1): print(result) ``` -------------------------------- ### Synchronous GraphQL Request with RequestsHTTPTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/requests.md Use this transport for synchronous GraphQL requests. Ensure the 'requests' library is installed. The transport can be configured with the GraphQL endpoint URL, SSL verification settings, and the number of retries for failed requests. ```python from gql import Client, gql from gql.transport.requests import RequestsHTTPTransport transport = RequestsHTTPTransport( url="https://countries.trevorblades.com/", verify=True, retries=3, ) client = Client(transport=transport, fetch_schema_from_transport=True) query = gql( """ query getContinents { continents { code name } } """ ) result = client.execute(query) print(result) ``` -------------------------------- ### Instantiate and Configure Fragment Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Instantiate a DSLFragment with a name, specify the GraphQL type it applies to using 'on', and add fields using 'select'. ```python name_and_appearances = DSLFragment("NameAndAppearances") name_and_appearances.on(ds.Character) name_and_appearances.select(ds.Character.name, ds.Character.appearsIn) ``` -------------------------------- ### gql-cli Usage Help Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Displays the help message for the gql-cli, outlining available arguments and options for executing GraphQL queries. ```bash usage: gql-cli [-h] [-V [VARIABLES ...]] [-H [HEADERS ...]] [--version] [-d | -v] [-o OPERATION_NAME] [--print-schema] [--schema-download [SCHEMA_DOWNLOAD ...]] [--execute-timeout EXECUTE_TIMEOUT] [--transport {auto,aiohttp,httpx,phoenix,websockets,aiohttp_websockets,appsync_http,appsync_websockets}] [--api-key API_KEY | --jwt JWT] server ``` -------------------------------- ### Provide Schema from File Source: https://github.com/graphql-python/gql/blob/master/docs/usage/validation.md Load a GraphQL schema from a .graphql file and pass it to the Client constructor. ```python with open('path/to/schema.graphql') as f: schema_str = f.read() client = Client(schema=schema_str) ``` -------------------------------- ### Basic Sync GraphQL Query Execution Source: https://github.com/graphql-python/gql/blob/master/README.md Demonstrates how to execute a basic GraphQL query using the sync client. Ensure no asyncio event loop is running, as this can cause conflicts. ```python from gql import Client, gql from gql.transport.aiohttp import AIOHTTPTransport # Select your transport with a defined url endpoint transport = AIOHTTPTransport(url="https://countries.trevorblades.com/") # Create a GraphQL client using the defined transport client = Client(transport=transport) # Provide a GraphQL query query = gql( """ query getContinents { continents { code name } } """ ) # Execute the query on the transport result = client.execute(query) print(result) ``` -------------------------------- ### Initialize DSLSchema Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Generate the root DSLSchema instance by passing the client's schema to DSLSchema. ```python ds = DSLSchema(client.schema) ``` -------------------------------- ### Run Full Test Suite Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Execute the complete test suite with coverage reporting. ```console pytest tests --cov=gql --cov-report=term-missing -vv ``` -------------------------------- ### Create Virtual Environment Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Use this command to create a virtual environment for development. ```console virtualenv gql-dev ``` -------------------------------- ### Execute Queries and Subscriptions with WebsocketsTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/websockets.md Demonstrates how to establish a WebSocket connection, execute a single query, and subscribe to real-time data using the WebsocketsTransport. Ensure the `async with Client` block is used to manage the connection lifecycle. ```python import asyncio import logging from gql import Client, gql from gql.transport.websockets import WebsocketsTransport logging.basicConfig(level=logging.INFO) async def main(): transport = WebsocketsTransport(url="wss://countries.trevorblades.com/graphql") # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with Client( transport=transport, fetch_schema_from_transport=True, ) as session: # Execute single query query = gql( """ query getContinents { continents { code name } } """ ) result = await session.execute(query) print(result) # Request subscription subscription = gql( """ subscription { somethingChanged { id } } """ ) async for result in session.subscribe(subscription): print(result) asyncio.run(main()) ``` -------------------------------- ### Execute Query from File Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Save a GraphQL query to a file and then execute it by piping the file content to gql-cli. ```bash $ echo 'query { continent(code:"AF") { name } }' > query.gql ``` ```bash $ cat query.gql | gql-cli wss://countries.trevorblades.com/graphql {"continent": {"name": "Africa"}} ``` -------------------------------- ### FastAPI Startup and Shutdown Events Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/async_permanent_session.md Configure FastAPI startup and shutdown events to manage the asynchronous GraphQL client's connection. This ensures the client is connected before requests are handled and properly closed when the application stops. ```python import logging from fastapi import FastAPI, HTTPException from fastapi.responses import HTMLResponse from gql import Client, gql from gql.client import ReconnectingAsyncClientSession from gql.transport.aiohttp import AIOHTTPTransport logging.basicConfig(level=logging.DEBUG) log = logging.getLogger(__name__) transport = AIOHTTPTransport(url="https://countries.trevorblades.com/graphql") client = Client(transport=transport) query = gql( """ query getContinentInfo($code: ID!) { continent(code:$code) { name code countries { name capital } } } """ ) app = FastAPI() @app.on_event("startup") async def startup_event(): print("Connecting to GraphQL backend") await client.connect_async(reconnecting=True) print("End of startup") @app.on_event("shutdown") async def shutdown_event(): print("Shutting down GraphQL permanent connection...") await client.close_async() print("Shutting down GraphQL permanent connection... done") ``` -------------------------------- ### Execute AppSync Subscriptions via gql-cli (WebSockets) Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Execute AppSync GraphQL subscriptions from the command line using gql-cli with the --transport appsync_websockets argument. Ensure the AWS_GRAPHQL_API_ENDPOINT environment variable is set. ```bash echo "subscription{onCreateMessage{message}}" | gql-cli $AWS_GRAPHQL_API_ENDPOINT --transport appsync_websockets ``` -------------------------------- ### Execute AppSync Mutations via gql-cli (HTTP) Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Run AppSync GraphQL mutations from the command line using gql-cli with the --transport appsync_http argument. Ensure the request is saved to a file and environment variables are set. ```bash # Put the request in a file $ echo 'mutation createMessage($message: String!) { createMessage(input: {message: $message}) { id message createdAt } }' > mutation.graphql # Execute the request using gql-cli with --transport appsync_http $ cat mutation.graphql | gql-cli $AWS_GRAPHQL_API_ENDPOINT --transport appsync_http -V message:"Hello world!" ``` -------------------------------- ### Asynchronous Subscription Execution Source: https://github.com/graphql-python/gql/blob/master/docs/usage/subscriptions.md Recommended for non-trivial tasks, this allows efficient concurrent queries and subscriptions. The `async with client` syntax manages the connection lifecycle. ```python import asyncio from gql import Client, gql from gql.transport.websockets import WebsocketsTransport async def main(): # Select your transport with a defined url endpoint transport = WebsocketsTransport(url='wss://your_server/graphql') # Create a GraphQL client using the defined transport client = Client(transport=transport) # Provide a GraphQL subscription query query = gql('\ subscription yourSubscription {\ ... } ') # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with client as session: # Then get the results using 'async for' async for result in session.subscribe(query): print (result) asyncio.run(main()) ``` -------------------------------- ### Inline Fragment with Shortcut Syntax Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Use the DSL shortcut syntax by passing '...' directly to the __call__ method for inline fragments. ```python query_with_inline_fragment = ds.Query.hero.args(episode=6).select( ds.Character.name, ds("...").on(ds.Human).select(ds.Human.homePlanet) ) ``` -------------------------------- ### Provide Schema from Python Object Source: https://github.com/graphql-python/gql/blob/master/docs/usage/validation.md Instantiate the Client with a GraphQLSchema object created using Python classes. ```python from .someSchema import SampleSchema # SampleSchema is an instance of GraphQLSchema client = Client(schema=SampleSchema) ``` -------------------------------- ### Handle GraphQL Subscriptions with AIOHTTPTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/aiohttp.md This snippet demonstrates how to handle GraphQL subscriptions using the multipart subscription protocol. It requires importing `Client`, `gql`, and `AIOHTTPTransport`, and then iterating over the results of `session.subscribe`. ```python import asyncio import logging from gql import Client, gql from gql.transport.aiohttp import AIOHTTPTransport logging.basicConfig(level=logging.INFO) async def main(): transport = AIOHTTPTransport(url="https://gql-book-server.fly.dev/graphql") # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with Client( transport=transport, ) as session: # Request subscription subscription = gql( """ subscription { book { title author } } """ ) # Subscribe and receive streaming updates async for result in session.subscribe(subscription): print(f"Received: {result}") asyncio.run(main()) ``` -------------------------------- ### Execute Query Against Local Schema Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/local_schema.md Instantiate a gql Client with a local schema and execute a query. This bypasses network transports and is ideal for testing. ```python from gql import gql, Client from .someSchema import SampleSchema client = Client(schema=SampleSchema) query = gql(''' { hello } ''') result = client.execute(query) ``` -------------------------------- ### Apply Directive to Mutation Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Demonstrates how to attach a directive to a mutation operation. ```python mutation = DSLMutation( ds.Mutation.createReview.args(episode=6, review={"stars": 5}).select( ds.Review.stars ) ).directives(ds("@customMutationDirective")) ``` -------------------------------- ### Console GraphQL Client with Async Session Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/async_permanent_session.md Create a console application that uses an asynchronous GraphQL client with a persistent session. This client connects to a GraphQL endpoint, allows users to query continent names by code, and handles potential exceptions during the process. ```python import asyncio import logging from typing import Optional from aioconsole import ainput from gql import Client, gql from gql.client import AsyncClientSession from gql.transport.aiohttp import AIOHTTPTransport logging.basicConfig(level=logging.INFO) GET_CONTINENT_NAME = """ query getContinentName ($code: ID!) { continent (code: $code) { name } } """ class GraphQLContinentClient: def __init__(self): self._client = Client( transport=AIOHTTPTransport(url="https://countries.trevorblades.com/") ) self._session: Optional[AsyncClientSession] = None self.get_continent_name_query = gql(GET_CONTINENT_NAME) async def connect(self): self._session = await self._client.connect_async(reconnecting=True) async def close(self): await self._client.close_async() async def get_continent_name(self, code): self.get_continent_name_query.variable_values = {"code": code} assert self._session is not None answer = await self._session.execute(self.get_continent_name_query) return answer.get("continent").get("name") # type: ignore async def main(): continent_client = GraphQLContinentClient() continent_codes = ["AF", "AN", "AS", "EU", "NA", "OC", "SA"] await continent_client.connect() while True: answer = await ainput("\nPlease enter a continent code or 'exit':") answer = answer.strip() if answer == "exit": break elif answer in continent_codes: try: continent_name = await continent_client.get_continent_name(answer) print(f"The continent name is {continent_name}\n") except Exception as exc: print(f"Received exception {exc} while trying to get continent name") else: print(f"Please enter a valid continent code from {continent_codes}") await continent_client.close() asyncio.run(main()) ``` -------------------------------- ### Create Conda Environment Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Create a new Conda environment with Python 3.8 for development. ```sh conda create -n gql-dev python=3.8 ``` -------------------------------- ### Interactive Query Execution Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Initiate an interactive session with gql-cli to execute GraphQL queries. Input queries directly in the terminal and press Ctrl-D to execute. ```bash $ gql-cli wss://countries.trevorblades.com/graphql --variables code:AF ``` -------------------------------- ### Meta-field with Shortcut Syntax Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Use the DSL shortcut syntax by passing the meta-field name string directly to the __call__ method. ```python query = ds.Query.hero.select( ds.Character.name, ds("__typename") ) ``` -------------------------------- ### Define GraphQL Requests for Batching Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/batching_requests.md Create a list of GraphQLRequest objects, each potentially including query, variable values, and operation name. Ensure your backend supports batch requests. ```python request1 = gql(""" query getContinents { continents { code name } } """ ) request2 = GraphQLRequest(""" query getContinentName ($code: ID!) { continent (code: $code) { name } } ", variable_values={ "code": "AF", }, ) requests = [request1, request2] ``` -------------------------------- ### Multiple File Uploads with gql Source: https://github.com/graphql-python/gql/blob/master/docs/usage/file_upload.md Upload multiple files by providing a list of FileVar instances to the mutation variables. Supports aiohttp, requests, and httpx transports. ```python from gql import client, gql, FileVar transport = AIOHTTPTransport(url='YOUR_URL') # Or transport = RequestsHTTPTransport(url='YOUR_URL') # Or transport = HTTPXTransport(url='YOUR_URL') # Or transport = HTTPXAsyncTransport(url='YOUR_URL') client = Client(transport=transport) query = gql(''' mutation($files: [Upload!]!) { multipleUpload(files: $files) { id } } ''') f1 = FileVar("YOUR_FILE_PATH_1") f2 = FileVar("YOUR_FILE_PATH_2") query.variable_values = {"files": [f1, f2]} result = client.execute(query, upload_files=True) ``` -------------------------------- ### Convert Operation to Executable Document Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Use the `dsl_gql` function to convert the defined DSL operation into a document that can be executed by the client or session. ```python query = dsl_gql( DSLQuery( ds.Query.hero.select( ds.Character.id, ds.Character.name, ds.Character.friends.select(ds.Character.name), ) ) ) result = client.execute(query) ``` -------------------------------- ### Execute Simple WebSocket Query Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Send a simple GraphQL query over WebSockets and display the JSON response. The query is piped to gql-cli. ```bash $ echo 'query { continent(code:"AF") { name } }' | gql-cli wss://countries.trevorblades.com/graphql {"continent": {"name": "Africa"}} ``` -------------------------------- ### Print GraphQL Schema Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Fetch and print the GraphQL schema from a given endpoint to a file using the --print-schema option. ```bash $ gql-cli https://countries.trevorblades.com/graphql --print-schema > schema.graphql ``` -------------------------------- ### Run Tox Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Execute tox tests within your activated virtual environment. ```console tox ``` -------------------------------- ### Create a Reconnecting Async Session Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/async_permanent_session.md Use `client.connect_async(reconnecting=True)` to establish a session that automatically attempts to reconnect if the transport connection is lost. Call `client.close_async()` for cleanup. ```python session = await client.connect_async(reconnecting=True) # You can run execute or subscribe method on this session result = await session.execute(query) # When you want the connection to close (for cleanup), # you call close_async await client.close_async() ``` -------------------------------- ### Synchronous Subscription Execution Source: https://github.com/graphql-python/gql/blob/master/docs/usage/subscriptions.md Use this for simple, non-concurrent subscription needs. Ensure the websockets transport is configured with the correct server URL. ```python from gql import Client, gql from gql.transport.websockets import WebsocketsTransport # Select your transport with a defined url endpoint transport = WebsocketsTransport(url='wss://your_server/graphql') # Create a GraphQL client using the defined transport client = Client(transport=transport) # Provide a GraphQL subscription query query = gql('\ subscription yourSubscription {\ ... } ') # Connect and subscribe to the results using a simple 'for' for result in client.subscribe(query): print (result) ``` -------------------------------- ### Running Async Code in IPython Environments Source: https://github.com/graphql-python/gql/blob/master/docs/usage/async_usage.md Provides alternative methods for running asynchronous code when an asyncio event loop is already active, such as in Jupyter or Spyder. Replace `asyncio.run(main())` with `await main()` or `loop.create_task(main())` depending on the environment. ```python await main() ``` ```python loop = asyncio.get_running_loop() loop.create_task(main()) ``` -------------------------------- ### Define Inline Fragment Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Instantiate DSLInlineFragment, specify the GraphQL type using 'on', and add fields with 'select'. ```python human_fragment = DSLInlineFragment() human_fragment.on(ds.Human) human_fragment.select(ds.Human.homePlanet) ``` -------------------------------- ### Create Document with Multiple Operations Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Construct a GraphQL document containing multiple operations, each with a distinct name. ```python query = dsl_gql( operation_name_1=DSLQuery( ... ), operation_name_2=DSLQuery( ... ), operation_name_3=DSLMutation( ... ), ) ``` -------------------------------- ### Apply Directive to Field Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Shows how to apply single or multiple directives to a specific field. ```python # Single directive on field ds.Query.hero.select( ds.Character.name.directives(ds("@customFieldDirective")) ) ``` ```python # Multiple directives on a field ds.Query.hero.select( ds.Character.appearsIn.directives( ds("@repeat").args(value="first"), ds("@repeat").args(value="second"), ds("@repeat").args(value="third"), ) ) ``` -------------------------------- ### Single File Upload with gql Source: https://github.com/graphql-python/gql/blob/master/docs/usage/file_upload.md Upload a single file by setting it as a variable value using FileVar and enabling upload_files. Supports aiohttp, requests, and httpx transports. ```python from gql import client, gql, FileVar transport = AIOHTTPTransport(url='YOUR_URL') # Or transport = RequestsHTTPTransport(url='YOUR_URL') # Or transport = HTTPXTransport(url='YOUR_URL') # Or transport = HTTPXAsyncTransport(url='YOUR_URL') client = Client(transport=transport) query = gql(''' mutation($file: Upload!) { singleUpload(file: $file) { id } } ''') query.variable_values = {"file": FileVar("YOUR_FILE_PATH")} result = client.execute(query, upload_files=True) ``` -------------------------------- ### Print AppSync Schema via gql-cli (HTTP) Source: https://github.com/graphql-python/gql/blob/master/docs/transports/appsync.md Retrieve the full GraphQL schema from an AppSync backend using gql-cli with the --transport appsync_http and --print-schema arguments. The schema will be saved to schema.graphql. ```bash $ gql-cli $AWS_GRAPHQL_API_ENDPOINT --transport appsync_http --print-schema > schema.graphql ``` -------------------------------- ### Execute GraphQL Query with Phoenix Channel Transport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/phoenix.md Demonstrates how to set up and use the PhoenixChannelWebsocketsTransport to execute a GraphQL query against an Absinthe backend. Ensure you replace 'YOUR_CHANNEL' and 'YOUR_URL' with your specific details. ```python import asyncio from gql import Client, gql from gql.transport.phoenix_channel_websockets import PhoenixChannelWebsocketsTransport async def main(): transport = PhoenixChannelWebsocketsTransport( channel_name="YOUR_CHANNEL", url="wss://YOUR_URL/graphql" ) # Using `async with` on the client will start a connection on the transport # and provide a `session` variable to execute queries on this connection async with Client(transport=transport) as session: # Execute single query query = gql( """ query yourQuery { ... } """ ) result = await session.execute(query) print(result) asyncio.run(main()) ``` -------------------------------- ### Instantiate DSLVariableDefinitions Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Instantiate DSLVariableDefinitions to prepare for variable arguments in GraphQL operations. ```python var = DSLVariableDefinitions() ``` -------------------------------- ### Apply Directive to Inline Fragment Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Demonstrates adding directives to an inline fragment. ```python query_with_directive = ds.Query.hero.args(episode=6).select( ds.Character.name, DSLInlineFragment().on(ds.Human).select(ds.Human.homePlanet).directives( ds("@customInlineFragmentDirective") ) ) ``` -------------------------------- ### Execute Simple HTTPS Query Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Send a simple GraphQL query over HTTPS and display the JSON response. The query is piped to gql-cli. ```bash $ echo 'query { continent(code:"AF") { name } }' | gql-cli https://countries.trevorblades.com {"continent": {"name": "Africa"}} ``` -------------------------------- ### Execute Batched Requests (Sync) Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/batching_requests.md Use the execute_batch method within a sync Client session to send a list of GraphQL requests as a single batch. This requires a compatible transport like RequestsHTTPTransport or HTTPXTransport. ```python transport = RequestsHTTPTransport(url=url) # Or transport = HTTPXTransport(url=url) with Client(transport=transport) as session: results = session.execute_batch(requests) result1 = results[0] result2 = results[1] ``` -------------------------------- ### Streaming Local Files with gql Source: https://github.com/graphql-python/gql/blob/master/docs/usage/file_upload.md Upload files without loading their entire content into memory by setting streaming=True in FileVar. This is only supported with the aiohttp transport. ```python transport = AIOHTTPTransport(url='YOUR_URL') client = Client(transport=transport) query = gql(''' mutation($file: Upload!) { singleUpload(file: $file) { id } } ''') f1 = FileVar( file_name='YOUR_FILE_PATH', streaming=True, ) query.variable_values = {"file": f1} result = client.execute(query, upload_files=True) ``` -------------------------------- ### Authenticate with HTTP Headers using AIOHTTPTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/aiohttp.md Configure authentication by passing an 'Authorization' header to the `AIOHTTPTransport` constructor. This is useful for token-based authentication. ```python transport = AIOHTTPTransport( url='https://SERVER_URL:SERVER_PORT/graphql', headers={'Authorization': 'token'} ) ``` -------------------------------- ### Enable GQL Debug Logging Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/logging.md Set the logging level to DEBUG to view messages exchanged between the client and server. This should be done at the beginning of your script. ```python import logging logging.basicConfig(level=logging.DEBUG) ``` -------------------------------- ### Execute Query with Variables Source: https://github.com/graphql-python/gql/blob/master/docs/gql-cli/intro.md Send a GraphQL query with variables using the --variables option. The query and variables are piped to gql-cli. ```bash $ echo 'query getContinent($code:ID!) { continent(code:$code) { name } }' | gql-cli https://countries.trevorblades.com --variables code:AF {"continent": {"name": "Africa"}} ``` -------------------------------- ### Run Online Tests Source: https://github.com/graphql-python/gql/blob/master/CONTRIBUTING.md Execute tests that require external online resources. ```console pytest tests --cov=gql --cov-report=term-missing --run-online -vv ``` -------------------------------- ### Generate a Query using DSL Schema Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Create a GraphQL query dynamically using the DSL module by selecting fields from the schema. This is equivalent to writing the query as a string. ```python ds = DSLSchema(StarWarsSchema) query = dsl_gql( DSLQuery( ds.Query.hero.select( ds.Character.id, ds.Character.name, ds.Character.friends.select(ds.Character.name), ) ) ) ``` ```python query = gql(""" query { hero { id name friends { name } } } """) ``` -------------------------------- ### Define Query with Variable and Directives using DSL Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md This snippet demonstrates how to define a GraphQL query with variable definitions and directives using the gql-python DSL. It shows the programmatic construction of a query that mirrors a standard GraphQL query string. ```python var.episode.directives(ds("@variableDefinition")) query.variable_definitions = var # Generate the document document = dsl_gql(character_fragment, query) ``` -------------------------------- ### Authenticate with HTTP Cookies using AIOHTTPTransport Source: https://github.com/graphql-python/gql/blob/master/docs/transports/aiohttp.md Manually set cookies for authentication by passing a 'cookies' dictionary to the `AIOHTTPTransport` constructor. Alternatively, use a cookie jar for managing cookies across connections, especially after a login mutation. ```python transport = AIOHTTPTransport(url=url, cookies={"cookie1": "val1"}) ``` ```python jar = aiohttp.CookieJar() transport = AIOHTTPTransport(url=url, client_session_args={'cookie_jar': jar}) ``` -------------------------------- ### Construct Complete Query Tree Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Build a complex query tree by nesting `select` calls to include fields from related objects, such as selecting friend names from the hero's friends list. ```python ds.Query.hero.select( ds.Character.id, ds.Character.name, ds.Character.friends.select(ds.Character.name), ) ``` -------------------------------- ### Define Mutation with Variable Arguments Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Construct a GraphQL mutation with variable arguments, select fields, assign variable definitions, and generate the query string. ```python var = DSLVariableDefinitions() op = DSLMutation( ds.Mutation.createReview.args(review=var.review, episode=var.episode).select( ds.Review.stars, ds.Review.commentary ) ) op.variable_definitions = var query = dsl_gql(op) ``` -------------------------------- ### Execute Batched Requests (Async) Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/batching_requests.md Execute a batch of GraphQL requests asynchronously using the execute_batch method within an async Client session. Compatible transports include AIOHTTPTransport or HTTPXAsyncTransport. ```python transport = AIOHTTPTransport(url=url) # Or transport = HTTPXAsyncTransport(url=url) async with Client(transport=transport) as session: results = await session.execute_batch(requests) result1 = results[0] result2 = results[1] ``` -------------------------------- ### Define Meta-field (__typename) Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Use DSLMetaField to include meta-fields like '__typename' in your query. ```python query = ds.Query.hero.select( ds.Character.name, DSLMetaField("__typename") ) ``` -------------------------------- ### Sequential Field Selection Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Fields can also be selected sequentially by first obtaining the DSLField instance and then calling `select` multiple times. ```python hero_query = ds.Query.hero hero_query.select(ds.Character.name) hero_query.select(ds.Character.id) ``` -------------------------------- ### Apply Directive to Variable Definition Source: https://github.com/graphql-python/gql/blob/master/docs/advanced/dsl_module.md Shows how to attach a directive to a specific variable definition. ```python var = DSLVariableDefinitions() var.episode.directives(ds("@customVariableDirective")) # Note: the directive is attached to the `.episode` variable definition (singular), # and not the `var` variable definitions (plural) holder. op = DSLQuery(ds.Query.hero.args(episode=var.episode).select(ds.Character.name)) op.variable_definitions = var ```