Asynchronise lets you turn your functions into fully coordinated, event-driven async generators that yield their results mid-step, allowing other functions (both synchronous and asynchronous, generators or non-generators) to use them immediately.

It's ideal if you need a small, tidy, async and sync-compatible framework for managing events with runtime type-checking. Or if you really love decorators.

Installing

pip install git+https://github.com/ckoshka/asynchronise

New example (working)

from _asynchronise_2 import Asynchronise
import random
asyncer = Asynchronise()
import asyncio
import typing
class Newspaper:
    def __init__(self, headline: str) -> None:
        self.headline = headline
        self.authors = []
    def __repr__(self) -> str:
        return f"Newspaper({self.headline})"

@asyncer.send
async def generate_newspaper():
    subject = ["Russia", "China", "The President", "H. P. Lovecraft", "The Moon"]
    verb = ["eats", "destroys", "kills", "saves"]
    obj = ["a banana", "a cat", "a dog", "a human", "a robot"]
    while True:
        yield Newspaper(f"{random.choice(subject)} {random.choice(verb)} {random.choice(obj)}"), "newspaper_no_authors"

def generate_authors():
    import time
    time.sleep(0.3)
    first_name_consonants = ["K", "P", "S", "T", "R", "L", "N", "M", "H", "W", "Y", "B", "G", "D", "J", "Z", "C", "Q", "X", "V", "F", "A", "E", "I", "O", "U"]
    first_name_syllable = ["agga", "iggy", "ogda", "edda"]
    last_name = str(random.randint(1, 9999))
    return f"{random.choice(first_name_consonants)}{random.choice(first_name_syllable)} {last_name}"

@asyncer.send # 'send' needs to be at the top
@asyncer.collect()
def add_author(newspaper_no_authors: Newspaper):
    author = generate_authors()
    yield author, "author"
    newspaper_no_authors.authors.append(author)
    yield newspaper_no_authors, "newspaper_with_authors"

@asyncer.collect({
    "newspaper_with_authors": lambda newspaper: newspaper.authors[0].startswith("K")
})
def print_newspaper(newspaper_with_authors: Newspaper):
    print(newspaper_with_authors)

@asyncer.collect()
def print_author(author: str):
    print(author)

async def run():
    i = 0
    async for newspaper in generate_newspaper():
        i += 1
        if i > 5:
            break
        pass
    await asyncio.sleep(2)

asyncio.run(run())

Old example

from asynchronise import Asynchronise

asy = Asynchronise()

@asy.send
async def do_expensive_multistep_stuff():
    flour: Ingredient[Flour] = await go_to_the_store()
    yield flour

    egg: Egg = await get_eggs(variety="scrambled")
    yield eggs, "scrambled"

    egg: Egg = await get_eggs(variety="raw")
    yield eggs, "raw"

    milk: Ingredient[Milk] = await get_milk()
    yield milk, "liquid"

@asy.collect({
    "egg": (Egg, "scrambled", lambda egg: egg.weight > 100)
})
def eat_egg(egg):
    # This won't block the event loop 
    time.sleep(10) 
    print("Yummy!")
    
@asy.send
@asy.collect({
    "orders": (int, "customer_order", None),
    "flour": (Flour, None, lambda flour: flour.type == "rye"),
    "liquid": (None, "liquid", None)
})
async def make_dough(orders, flour, liquid):
    # I don't actually understand how making bread works, sorry to any bakers out there.
    for order_num in range(orders):
        await liquid.stir()
        dough = await flour.mix(liquid)
        yield dough, order_num