caarmen/swapi-graphql-client.py

## swapi-graphql-client.py
"""
Exploration of a graphql client in Python
"""
import asyncio
from dataclasses import dataclass
from typing import Generator

from gql import gql
from gql.client import Client, AsyncClientSession
from gql.transport.aiohttp import AIOHTTPTransport


def _create_query(after_cursor: str = None) -> str:
    after = '' if after_cursor is None else f'after: "{after_cursor}"'
    return f'''
query {{
    allPeople(first:10, {after}) {{
        pageInfo {{
            endCursor
            hasNextPage
        }}
        edges {{
            cursor
            node {{
                name
                homeworld {{
                    name
                }}
            }}
        }}
    }}
}}
    '''


@dataclass
class SwEntry:
    """
    One entry in the sw dataset
    """
    person_name: str
    homeworld_name: str


def _person_to_sw_entry(person: dict) -> SwEntry:
    return SwEntry(
        person_name=person["name"],
        homeworld_name=person["homeworld"]["name"]
    )


def _people_edges_to_sw_entries(people_edges: list[dict]) -> list[SwEntry]:
    return [_person_to_sw_entry(people_edge["node"]) for people_edge in people_edges]


async def _result_generator(session: AsyncClientSession) -> Generator[list[SwEntry], None, None]:
    page_info = {}
    while page_info.get("hasNextPage", True):
        # Get the current page data
        query = gql(_create_query(page_info.get("endCursor")))
        data = await session.execute(query)
        people_connection = data["allPeople"]
        yield _people_edges_to_sw_entries(people_connection["edges"])

        # Prepare the next page query
        page_info = people_connection["pageInfo"]


def _sw_entry_to_display(sw_entry: SwEntry) -> str:
    return f"{sw_entry.person_name}: {sw_entry.homeworld_name}"


async def main():
    """
    Print data from the swapi graphql endpoint
    """
    transport = AIOHTTPTransport(url="https://swapi-graphql.netlify.app/.netlify/functions/index")
    async with Client(
            transport=transport,
            fetch_schema_from_transport=True,
            serialize_variables=True,
            parse_results=True
    ) as session:
        async for sw_entries in _result_generator(session):
            sw_entries_display_strings = [_sw_entry_to_display(sw_entry) for sw_entry in sw_entries]
            print("\n".join(sw_entries_display_strings))


asyncio.run(main())
	"""
	Exploration of a graphql client in Python
	"""
	import asyncio
	from dataclasses import dataclass
	from typing import Generator

	from gql import gql
	from gql.client import Client, AsyncClientSession
	from gql.transport.aiohttp import AIOHTTPTransport


	def _create_query(after_cursor: str = None) -> str:
	after = '' if after_cursor is None else f'after: "{after_cursor}"'
	return f'''
	query {{
	allPeople(first:10, {after}) {{
	pageInfo {{
	endCursor
	hasNextPage
	}}
	edges {{
	cursor
	node {{
	name
	homeworld {{
	name
	}}
	}}
	}}
	}}
	}}
	'''


	@dataclass
	class SwEntry:
	"""
	One entry in the sw dataset
	"""
	person_name: str
	homeworld_name: str


	def _person_to_sw_entry(person: dict) -> SwEntry:
	return SwEntry(
	person_name=person["name"],
	homeworld_name=person["homeworld"]["name"]
	)


	def _people_edges_to_sw_entries(people_edges: list[dict]) -> list[SwEntry]:
	return [_person_to_sw_entry(people_edge["node"]) for people_edge in people_edges]


	async def _result_generator(session: AsyncClientSession) -> Generator[list[SwEntry], None, None]:
	page_info = {}
	while page_info.get("hasNextPage", True):
	# Get the current page data
	query = gql(_create_query(page_info.get("endCursor")))
	data = await session.execute(query)
	people_connection = data["allPeople"]
	yield _people_edges_to_sw_entries(people_connection["edges"])

	# Prepare the next page query
	page_info = people_connection["pageInfo"]


	def _sw_entry_to_display(sw_entry: SwEntry) -> str:
	return f"{sw_entry.person_name}: {sw_entry.homeworld_name}"


	async def main():
	"""
	Print data from the swapi graphql endpoint
	"""
	transport = AIOHTTPTransport(url="https://swapi-graphql.netlify.app/.netlify/functions/index")
	async with Client(
	transport=transport,
	fetch_schema_from_transport=True,
	serialize_variables=True,
	parse_results=True
	) as session:
	async for sw_entries in _result_generator(session):
	sw_entries_display_strings = [_sw_entry_to_display(sw_entry) for sw_entry in sw_entries]
	print("\n".join(sw_entries_display_strings))


	asyncio.run(main())