Async Transformers

Transformers can be also created from coroutines, but, in this case, you need to use the @async_transformer decorator. For example, consider the transformer below that fetches data from an external http server:

import httpx
from gloe import async_transformer

async def fetch_data(resource: str) -> httpx.Response:
    async with httpx.AsyncClient() as client:
        r = await client.get(f'{BASE_URL}/{resource}')
    return r

The @async_transformer decorator converts the fetch_data function into an instance of the AsyncTransformer class.

To call this transformer, we need to use the await syntax, just like a normal coroutine:

await fetch_data('users')

Async Pipelines

An async transformer can be composed with other async transformers or with other sync transformers (normal transformers). For example:

async def get_user_by_id(user_id: int) -> User: ...

def extract_user_roles(user: User) -> list[Role]: ...

get_user_roles = get_user_by_id >> extract_user_roles

When a pipeline has at least one async transformer, the entire pipeline becomes async, so we must call it using the await statement:

await get_user_roles(user_id)

Alternatively, we can start the pipeline with a sync transformer and append an async transformer to it:

def extract_user_id(request: Request) -> int: ...

get_user = extract_user_id >> get_user_by_id

Briefly, you can mix async and sync transformers together without worrying about its concurrent nature, but it is still a transformer, so you’ll still have to make sure the typing is correct.

Partial Async Transformers

Just like in sync transformers, you can create partial transformers with an async behavior. It can be done using the @partial_async_transformer:

from gloe import partial_async_transformer

async def get_users_by_role(role: str, page: int = 0, page_size: int = 10) -> Page[User]:
    "Get users with a specific role. The response is paginated."