anyio-ext

A collection of helpful utilities for anyio.

Features

• Powerful, easy-to-use caching decorator
• Typed asyncio.gather implementation
• Generic Queue and Stack primitives

Installation

$ pip install anyio-ext

Getting started

gather() is similar to asyncio.gather():

from anyio import sleep
from anyio_ext import gather

results = await gather(*[sleep(i) for i in range(3)])
print(results)  # (None, None, None)

Caching is implemented with a decorator:

from anyio_ext import cached

@cached(ttl=60)
async def my_task() -> int: ...

Queues

Queue and stack implementations work similarly:

from anyio_ext import Queue, Stack

queue = Queue[int](max_size=32)
stack = Stack[int](max_size=32)

await queue.push(1)
await queue.push(2)
print(await queue.pop())  # 1
await stack.push(1)
await stack.push(2)
print(await stack.pop())  # 2

Advanced caching

Cache keys are generated by hashing arguments. You can exclude non-serializable arguments from cache key construction:

from sqlalchemy.ext.asyncio import AsyncSession

@cached(ttl=60, exclude={"session"})
async def my_task(session: AsyncSession) -> int: ...

You can also customize which parts of arguments get hashed:

@cached(
    ttl=60,
    key_fns={
        # hash just the ID, not the entire model
        "user": lambda u: u.id,
        # hash a couple relevant fields
        "message": lambda m: (m.type, m.timestamp),
    },
)
async def my_task(user: User, message: Message) -> int: ...

You can easily invalidate keys by passing the same arguments:

@cached(ttl=60)
async def my_task(time: int) -> int: ...

await my_task.invalidate(3)

Simple cache statistics are available:

print(my_task.hits, my_task.misses, len(my_task))  # 999 1 1

Methods can also be cached:

class MyClass:
    @cached(ttl=60)
    async def my_task(self, time: int) -> int: ...

instance = MyClass()
await instance.my_task(3)
instance.my_task.invalidate(3)
print(MyClass.my_task.hits, MyClass.my_task.misses, len(MyClass.my_task))  # 0 1 0

Note that statistics are on a per-class basis, but calls to my_task() are on a per-instance basis (even though behind the scenes, the cache is shared across all instances). self is one of the parameters used for caching by default, but it won't cause memory leaks as keys don't store references to arguments.