Deferring code

quattro allows you to defer runing async or sync functions until the end of a coroutine’s execution. This keeps code indentation reasonable (and hence more readable) when using context managers, which are a staple of robust resource management.

This functionality is loosely inspired by the defer statement in the Go programming language.

When and where to use

• Use with context managers (sync and async) that roughly match the coroutine scope, and don’t require error handling on this layer (but, for example, on higher layers).

• Use combined with contextlib.aclosing() to ensure any async generators are properly closed.

quattro.Deferrer

Deferrer is a helper class for deferring functions and coroutines.

Deferrer can be applied to a coroutine function in the following way:

0. Let’s start with a simple coroutine function:

async def my_coroutine_function(a: int) -> str:
    await sleep(1)
    return str(a)

1. Add a parameter to your coroutine function. It should be the first parameter, and not keyword-only. It can be named whatever you like, but we recommend defer for consistency. The type hint is completely optional.

from quattro import Deferrer

async def my_coroutine_function(defer: Deferrer, a: int) -> str:
    await sleep(1)
    return str(a)

2. Apply the Deferrer.enable() decorator to the coroutine function. The decorator is implemented as a class method on the Deferrer class to minimize the number of names you need to import.

from quattro import Deferrer

@Deferrer.enable
async def my_coroutine_function(defer: Deferrer, a: int) -> str:
    await sleep(1)
    return str(a)

3. You’re done! You can use the defer parameter in the coroutine body. Since Deferrer implements __call__, you can call it directly:

from quattro import Deferrer

@Deferrer.enable
async def my_coroutine_function(defer: Deferrer, a: int) -> str:
    taskgroup = defer(TaskGroup())
    taskgroup.create_task(some_other_coroutine())
    await sleep(1)
    return str(a)

Deferrer is a subclass of Python’s AsyncExitStack, and so supports all of its methods.

from quattro import Deferrer

@Deferrer.enable
async def my_coroutine_function(defer: Deferrer, a: int) -> str:
    defer.push_async_callback(some_other_coroutine, a)
    return str(a)

quattro.defer

quattro.defer() is a more magical and more succint version of quattro.Deferrer.

When and where to use

• When you want to defer with a little less typing, more readability and less type-safety.

• When you need to have a clean function signature, maybe for use with a CLI library that requires it.

Apply the defer.enable decorator to a coroutine function, and then call defer() inside.

from quattro import defer

@defer.enable
async def my_coroutine_function(a: int) -> str:
    defer(TaskGroup())
    await sleep(1)
    return str(a)

quattro.defer also supports AsyncExitStacks enter_context method, for dealing with non-async context managers.

@defer.enable
async def my_coroutine_function(a: int) -> str:
    defer.enter_context(fail_after(5))  # `fail_after` is a sync context manager.
    await sleep(1)
    return str(a)

Warning

Do not mix defer() and Deferrer in the same coroutine function; pick one or the other.