API reference

Context

A context allows to share values. When a shared value is put in a context, it can be borrowed by calling await context.get(value_type), where value_type is the type of the desired value.

__init__(*, close_timeout=None)

Parameters:

Name	Type	Description	Default
close_timeout	float | None	The timeout to use when closing the context.	None

aclose(*, timeout=None, _exc_type=None, _exc_value=None, _exc_tb=None) async

Close the context, after all shared values that were borrowed have been dropped. The shared values will be torn down, if applicable.

Parameters:

Name	Type	Description	Default
timeout	float | None	The time to wait for all shared values to be freed.	None

Raises:

Type	Description
TimeoutError	If the context could not be closed in time.

add_teardown_callback(teardown_callback)

Register a callback that will be called at context teardown. The callbacks will be called in the inverse order than they were added.

Parameters:

Name	Type	Description	Default
teardown_callback	Callable[..., Any] | Callable[..., Awaitable[Any]]	The callback to add.	required

get(value_type, timeout=float('inf')) async

Get a value from the context, with the given type. The value will be returned if/when it is put in the context and when it accepts to be borrowed (borrowing can be limited with a maximum number of borrowers).

Parameters:

Name	Type	Description	Default
value_type	type[T]	The type of the value to get.	required
timeout	float	The time to wait to get the value.	float('inf')

Returns:

Type	Description
Value[T]	The borrowed Value.

Raises:

Type	Description
TimeoutError	If the value could not be borrowed in time.

put(value, types=None, max_borrowers=float('inf'), manage=False, teardown_callback=None, shared_value=None)

Put a value in the context so that it can be shared.

Parameters:

Name	Type	Description	Default
value	T	The value to put in the context.	required
types	Iterable | Any | None	The type(s) to register the value as. If not provided, the value type will be used.	None
max_borrowers	float	The number of times the shared value can be borrowed at the same time.	float('inf')
manage	bool	Whether to use the (async) context manager of the value for setup/teardown.	False
teardown_callback	Callable[..., Any] | Callable[..., Awaitable[Any]] | None	An optional callback to call when the context is closed.	None

Returns:

Type	Description
SharedValue[T]	The shared value.

Module

A module allows to:

• run services,
• share those services with other modules,
• request services from other modules.

The services are represented by values that can be published by producers and borrowed by consumers. Consumers notify producers that their services are not used anymore by dropping the corresponding borrowed values. Producers are responsible for tearing down their services when stopping the application.

Modules can be configured through their __init__ method's keyword arguments. Modules have three phases:

• prepare: called before the "start" phase.
• start: called before running the application.
• stop: called when shutting down the application.

modules property

Returns:

Type	Description
dict[str, Module]	The modules added by the current module, as a dict of module name to module instance.

name property

Returns:

Type	Description
str	The module's name.

parent property writable

Returns:

Type	Description
Module | None	The module's parent, unless this is the root module which has no parent.

path property

Returns:

Type	Description
str	The module's path, as a period-separated sequence of module names.

started property

Returns:

Type	Description
Event	An Event that is set when the module has started.

__init__(name, prepare_timeout=1, start_timeout=1, stop_timeout=1)

Parameters:

Name	Type	Description	Default
name	str	The name to give to the module.	required
prepare_timeout	float	The time to wait (in seconds) for the "prepare" phase to complete.	1
start_timeout	float	The time to wait (in seconds) for the "start" phase to complete.	1
stop_timeout	float	The time to wait (in seconds) for the "stop" phase to complete.	1

add_module(module_type, name, **config)

Add a module as a child of the current module.

Parameters:

Name	Type	Description	Default
module_type	type['Module'] | str	A Module type or a string pointing to a module type.	required
name	str	The name to give to the module.	required
config		The module configuration.	{}

add_teardown_callback(teardown_callback)

Register a callback that will be called when stopping the module. The callbacks will be called in the inverse order than they were added.

Parameters:

Name	Type	Description	Default
teardown_callback	Callable[..., Any] | Callable[..., Awaitable[Any]]	The callback to add.	required

all_freed() async

Wait for all published values to be freed, meaning that all borrowers have dropped their values.

done()

Notify that the current phase is done. This is especially useful when launching background tasks, as otherwise the current phase would not complete:

from anyio import create_task_group
from fps import Module

class MyModule(Module):
    async def start(self):
        async with create_task_group() as tg:
            tg.start_toon(my_async_func)
            tg.start_toon(other_async_func)
            self.done()

drop(value)

Drop a borrowed value, meaning that this module doesn't use it anymore.

Parameters:

Name	Type	Description	Default
value	Any	The value to drop.	required

drop_all()

Drop all borrowed values.

exit_app()

Force the application to exit. This can be called from any module.

freed(value) async

Wait for a published value to be free, meaning that all borrowers have dropped the value.

Parameters:

Name	Type	Description	Default
value	Any	The value to be freed.	required

get(value_type, timeout=float('inf')) async

Borrow a value from the current module's context or its parent's (if any).

Parameters:

Name	Type	Description	Default
value_type	type[T_Value]	The type of the value to borrow.	required
timeout	float	The time to wait for the value to be published.	float('inf')

Returns:

Type	Description
T_Value	The borrowed value.

prepare() async

The "prepare" phase occurs before the "start" phase.

put(value, types=None, max_borrowers=float('inf'), teardown_callback=None, manage=False)

Publish a value in the current module context and its parent's (if any).

Parameters:

Name	Type	Description	Default
value	T_Value	The value to publish.	required
types	Iterable | Any | None	The type(s) to publish the value as. If not provided, the type is inferred from the value.	None
max_borrowers	float	The maximum number of simultaneous borrowers of the published value.	float('inf')
teardown_callback	Callable[..., Any] | Callable[..., Awaitable[Any]] | None	A callback to call when the value is torn down.	None
manage	bool	Whether to use the (async) context manager of the value for its setup/teardown.	False

run(backend='asyncio')

Run the root module.

Parameters:

Name	Type	Description	Default
backend	str	The backend used to run ("asyncio" or "trio").	'asyncio'

start() async

The "start" phase occurs after the "prepare" phase. This is usually where services are started and published as values, and other services are requested and borrowed as values.

stop() async

The "stop" phase occurs when the application is torn down.

SharedValue

Bases: Generic[T]

A value that can be shared with so-called borrowers. A borrower borrows a shared value by calling await shared_value.get(), which returns a Value. The shared value can be borrowed any number of times at the same time, unless specified by max_borrowers. All borrowers must drop their Value before the shared value can be closed. The shared value can be closed explicitly by calling await shared_value.aclose(), or by using an async context manager.

__init__(value, max_borrowers=float('inf'), manage=False, teardown_callback=None, close_timeout=None)

Parameters:

Name	Type	Description	Default
value	T	The inner value that is shared.	required
max_borrowers	float	The number of times the shared value can be borrowed at the same time.	float('inf')
manage	bool	Whether to use the (async) context manager of the inner value for setup/teardown.	False
teardown_callback	Callable[..., Any] | Callable[..., Awaitable[Any]] | None	The callback to call when closing the shared value.	None
close_timeout	float | None	The timeout to use when closing the shared value.	None

aclose(*, timeout=None, _exc_type=None, _exc_value=None, _exc_tb=None) async

Wait for all borrowers to drop their value, and tear down the shared value.

Parameters:

Name	Type	Description	Default
timeout	float | None	The time to wait for all borrowers to drop their value.	None

Raises:

Type	Description
TimeoutError	If the shared value could not be closed in time.

freed(timeout=float('inf')) async

Wait for all borrowers to drop their value.

Parameters:

Name	Type	Description	Default
timeout	float	The time to wait for all borrowers to drop their value.	float('inf')

Raises:

Type	Description
TimeoutError	If the shared value was not freed in time.

get(timeout=float('inf')) async

Borrow the shared value.

Parameters:

Name	Type	Description	Default
timeout	float	The time to wait for the value to be dropped.	float('inf')

Returns:

Type	Description
Value	The borrowed value.

Raises:

Type	Description
TimeoutError	If the value could not be borrowed in time.

Value

Bases: Generic[T]

A Value can be obtained from a shared value by calling await shared_value.get(), and can be dropped by calling value.drop(). The inner value can be accessed by calling value.unwrap(), unless it was already dropped.

__init__(shared_value)

Parameters:

Name	Type	Description	Default
shared_value	SharedValue[T]	The shared value this Value refers to.	required

drop()

Drop the value.

unwrap()

Get the inner value that is shared.

Raises:

Type	Description
RuntimeError	If the value was already dropped.

Returns:

Type	Description
T	The inner value.