asyncgraphs 0.0.4

No project description provided

AsyncGraphs

---

AsyncGraphs is a tiny ETL framework that leverages asyncio to make the execution concurrent whilst blocked on I/O.

Source: https://github.com/SamVermeulen42/asyncgraphs

Documentation: https://samvermeulen42.github.io/asyncgraphs/

---

Features

• Typed
• Simple concurrency based on asyncio
• Easy construction of ETL graphs

Installation

pip install asyncgraphs

Example

The following example prints random Pokémon and the games they appear in.

It does this every 10 seconds and uses PokéApi.

import aiohttp
from asyncgraphs import Graph, run
import asyncio
from functools import partial
from random import randint
from typing import Dict, Any


async def random_pokemon_id():
    while True:
        yield randint(1, 151)
        await asyncio.sleep(10)

async def get_pokemon_info(session: aiohttp.ClientSession, pokemon_id: int) -> Dict[str, Any]:
    pokemon_url = f"https://pokeapi.co/api/v2/pokemon/{pokemon_id}"
    async with session.get(pokemon_url) as response:
        yield await response.json()
                
def format_pokemon(pokemon_info: Dict[str, Any]) -> str:
    name = pokemon_info["name"]
    versions = (game['version']['name'] for game in pokemon_info['game_indices'])
    return f"{name}: {', '.join(versions)}"

async def main():
    async with aiohttp.ClientSession() as session:
        g = Graph()
        g | random_pokemon_id() | partial(get_pokemon_info, session) | format_pokemon | print
        await run(g)

asyncio.run(main())