import configparser
import logging
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
from gql.transport.exceptions import TransportQueryError
from graphql.error.graphql_error import GraphQLError
from asyncio.exceptions import TimeoutError
# Configure logging
log_handler = logging.StreamHandler(sys.stdout)
log_handler.setLevel(logging.DEBUG)
log_format = logging.Formatter("%(levelname)s %(asctime)s %(message)s")
log_handler.setFormatter(log_format)
logger = logging.getLogger(__name__)
logger.addHandler(log_handler)
logger.setLevel(logging.INFO)
# Load the config file values
config = configparser.ConfigParser()
config.read("config.ini")
# Ghostwriter API URL and variables
GHOSTWRITER_API_URL = f"{config['ghostwriter']['gw_url'].strip('/')}/v1/graphql"
GHOSTWRITER_TOKEN = config['ghostwriter']['api_token']
try:
# Define some queries or mutations as `gql()` objects here
whoami_query = gql(
"""
query Whoami {
whoami {
username role expires
}
}
"""
)
# Configure the GQL transport
headers = {"Authorization": f"Bearer {GHOSTWRITER_TOKEN}"}
transport = AIOHTTPTransport(url=GHOSTWRITER_API_URL, headers=headers)
authenticated_client = Client(transport=transport, fetch_schema_from_transport=True)
# Test the token with a `whois` query
result = authenticated_client.execute(whoami_query)
logger.info(f"Authenticated as {result['whoami']['username']}")
# Execute queries and mutation with `authenticated_client.execute()` here
except TimeoutError:
# Do something...
pass
except TransportQueryError as e:
# Do something...
pass
except GraphQLError as e:
# Do something...
pass