Async

Icontract supports both adding sync contracts to coroutine functions as well as enforcing async conditions (and capturing async snapshots).

You simply define your conditions as decorators of a coroutine function:

import icontract

@icontract.require(lambda x: x > 0)
@icontract.ensure(lambda x, result: x < result)
async def do_something(x: int) -> int:
    ...

Special Considerations

Async conditions. If you want to enforce async conditions, the function also needs to be defined as async:

import icontract

async def has_author(author_id: str) -> bool:
    ...

@icontract.ensure(has_author)
async def upsert_author(name: str) -> str:
    ...

It is not possible to add an async condition to a sync function. Doing so will raise a ValueError at runtime. The reason behind this limitation is that the wrapper around the function would need to be made async, which would break the code calling the original function and expecting it to be synchronous.

Invariants. As invariants need to wrap dunder methods, including __init__, their conditions can not be async, as most dunder methods need to be synchronous methods, and wrapping them with async code would break that constraint. You can, of course, use synchronous invariants on async method functions without problems.

No async lambda. Another practical limitation is that Python does not support async lambda (see this Python issue), so defining async conditions (and snapshots) is indeed tedious (see the next section). Please consider asking for async lambdas on python-ideas mailing list to give the issue some visibility.

Coroutine as condition result. If the condition returns a coroutine, the coroutine will be awaited before it is evaluated for truthiness.

This means in practice that you can work around no-async lambda limitation applying coroutine functions on your condition arguments (which in turn makes the condition result in a coroutine).

For example:

async def some_condition(a: float, b: float) -> bool:
    ...

@icontract.require(lambda x: some_condition(a=x, b=x**2))
async def some_func(x: float) -> None:
    ...

A big fraction of contracts on sequences require an all operation to check that all the item of a sequence are True. Unfortunately, all does not automatically operate on a sequence of Awaitables, but the library asyncstdlib comes in very handy:

import asyncstdlib as a

Here is a practical example that uses asyncstdlib.map, asyncstdlib.all and asyncstdlib.await_each:

import asyncstdlib as a

async def has_author(identifier: str) -> bool:
    ...

async def has_category(category: str) -> bool:
    ...

@dataclasses.dataclass
class Book:
    identifier: str
    author: str

@icontract.require(lambda categories: a.map(has_category, categories))
@icontract.ensure(
    lambda result: a.all(a.await_each(has_author(book.author) for book in result)))
async def list_books(categories: List[str]) -> List[Book]:
    ...

Coroutines have side effects. If the condition of a contract returns a coroutine, the condition can not be re-computed upon the violation to produce an informative violation message. This means that you need to specify an explicit error which should be raised on contract violation.

For example:

async def some_condition() -> bool:
    ...

@icontract.require(
    lambda: some_condition(),
    error=lambda: icontract.ViolationError("Something went wrong."))

If you do not specify the error, and the condition returns a coroutine, the decorator will raise a ValueError at re-computation time.