`faust.app`¶

class faust.app.App(id: str, *, monitor: faust.sensors.monitor.Monitor = None, config_source: Any = None, loop: asyncio.events.AbstractEventLoop = None, **options) → None[source]¶

Faust Application.

Keyword Arguments:
Parameters:	id (str) – Application ID.
	loop (asyncio.AbstractEventLoop) – optional event loop to use.

See also

faust.channels.Channel.

Return type:	`ChannelT`[]

agent(channel: Union[str, faust.types.channels.ChannelT] = None, *, name: str = None, concurrency: int = 1, supervisor_strategy: Type[mode.types.supervisors.SupervisorStrategyT] = None, sink: Iterable[Union[_ForwardRef('AgentT'), faust.types.channels.ChannelT, typing.Callable[[typing.Any], typing.Union[typing.Awaitable, NoneType]]]] = None, isolated_partitions: bool = False, **kwargs) → Callable[Callable[Union[typing.AsyncIterator, faust.types.streams.StreamT], Union[typing.Awaitable, typing.AsyncIterable]], faust.types.agents.AgentT][source]¶

Create Agent from async def function.

It can be a regular async function:

@app.agent()
async def my_agent(stream):
    async for number in stream:
        print(f'Received: {number!r}')

Or it can be an async iterator that yields values. These values can be used as the reply in an RPC-style call, or for sinks: callbacks that forward events to other agents/topics/statsd, and so on:

@app.agent(sink=[log_topic])
async def my_agent(requests):
    async for number in requests:
        yield number * 2

Return type:	`Callable`[[`Callable`[[`Union`[`AsyncIterator`[+T_co], `StreamT`[+T_co]]], `Union`[`Awaitable`[+T_co], `AsyncIterable`[+T_co]]]], `AgentT`[]]

actor(channel: Union[str, faust.types.channels.ChannelT] = None, *, name: str = None, concurrency: int = 1, supervisor_strategy: Type[mode.types.supervisors.SupervisorStrategyT] = None, sink: Iterable[Union[_ForwardRef('AgentT'), faust.types.channels.ChannelT, typing.Callable[[typing.Any], typing.Union[typing.Awaitable, NoneType]]]] = None, isolated_partitions: bool = False, **kwargs) → Callable[Callable[Union[typing.AsyncIterator, faust.types.streams.StreamT], Union[typing.Awaitable, typing.AsyncIterable]], faust.types.agents.AgentT]¶

Create Agent from async def function.

It can be a regular async function:

@app.agent()
async def my_agent(stream):
    async for number in stream:
        print(f'Received: {number!r}')

Or it can be an async iterator that yields values. These values can be used as the reply in an RPC-style call, or for sinks: callbacks that forward events to other agents/topics/statsd, and so on:

@app.agent(sink=[log_topic])
async def my_agent(requests):
    async for number in requests:
        yield number * 2

Return type:	`Callable`[[`Callable`[[`Union`[`AsyncIterator`[+T_co], `StreamT`[+T_co]]], `Union`[`Awaitable`[+T_co], `AsyncIterable`[+T_co]]]], `AgentT`[]]

task(fun: Union[typing.Callable[[_ForwardRef('AppT')], typing.Awaitable], typing.Callable[[], typing.Awaitable]]) → Union[typing.Callable[[_ForwardRef('AppT')], typing.Awaitable], typing.Callable[[], typing.Awaitable]][source]¶

Define an async def function to be started with the app.

This is like timer() but a one-shot task only executed at worker startup (after recovery and the worker is fully ready for operation).

The function may take zero, or one argument. If the target function takes an argument, the app argument is passed:

>>> @app.task
>>> async def on_startup(app):
...    print('STARTING UP: %r' % (app,))

Nullary functions are also supported:

>>> @app.task
>>> async def on_startup():
...     print('STARTING UP')

Return type:	`Union`[`Callable`[[`AppT`[]], `Awaitable`[+T_co]], `Callable`[[], `Awaitable`[+T_co]]]

timer(interval: Union[datetime.timedelta, float, str], on_leader: bool = False) → Callable[source]¶

Define an async def function to be run at periodic intervals.

Like task(), but executes periodically until the worker is shut down.

This decorator takes an async function and adds it to a list of timers started with the app.

Parameters:	interval (Seconds) – How often the timer executes in seconds. on_leader (bool) – Should the timer only run on the leader?

Example

>>> @app.timer(interval=10.0)
>>> async def every_10_seconds():
...     print('TEN SECONDS JUST PASSED')

>>> app.timer(interval=5.0, on_leader=True)
>>> async def every_5_seconds():
...     print('FIVE SECONDS JUST PASSED. ALSO, I AM THE LEADER!')

Return type:	`Callable`

service(cls: Type[mode.types.services.ServiceT]) → Type[mode.types.services.ServiceT][source]¶

Decorate mode.Service to be started with the app.

Examples

from mode import Service

@app.service
class Foo(Service):
    ...

Return type:	`Type`[`ServiceT`[]]

is_leader() → bool[source]¶

Return type:	`bool`

stream(channel: Union[typing.AsyncIterable, typing.Iterable], beacon: mode.utils.types.trees.NodeT = None, **kwargs) → faust.types.streams.StreamT[source]¶

Create new stream from channel/topic/iterable/async iterable.

Parameters:	channel (`Union`[`AsyncIterable`[+T_co], `Iterable`[+T_co]]) – Iterable to stream over (async or non-async). kwargs (`Any`) – See `Stream`.
Return type:	`StreamT`[+T_co]
Returns:	to iterate over events in the stream.
Return type:	faust.Stream

Table(name: str, *, default: Callable[Any] = None, window: faust.types.windows.WindowT = None, partitions: int = None, help: str = None, **kwargs) → faust.types.tables.TableT[source]¶

Define new table.

Parameters:	name (`str`) – Name used for table, note that two tables living in the same application cannot have the same name. default (`Optional`[`Callable`[[], `Any`]]) – A callable, or type that will return a default value for keys missing in this table. window (`Optional`[`WindowT`]) – A windowing strategy to wrap this window in.

Examples

>>> table = app.Table('user_to_amount', default=int)
>>> table['George']
0
>>> table['Elaine'] += 1
>>> table['Elaine'] += 1
>>> table['Elaine']
2

Return type:	`TableT`[]

page(path: str, *, base: Type[faust.web.views.View] = <class 'faust.web.views.View'>) → Callable[Union[typing.Type[faust.types.web.View], typing.Callable[[faust.types.web.View, faust.types.web.Request], typing.Awaitable[faust.types.web.Response]]], Type[faust.web.views.Site]][source]¶

Return type:	`Callable`[[`Union`[`Type`[`View`], `Callable`[[`View`, `Request`], `Awaitable`[`Response`]]]], `Type`[`Site`]]

table_route(table: faust.types.tables.CollectionT, shard_param: str) → Callable[Callable[[faust.types.web.View, faust.types.web.Request], Awaitable[faust.types.web.Response]], Callable[[faust.types.web.View, faust.types.web.Request], Awaitable[faust.types.web.Response]]][source]¶

Return type:	`Callable`[[`Callable`[[`View`, `Request`], `Awaitable`[`Response`]]], `Callable`[[`View`, `Request`], `Awaitable`[`Response`]]]

command(*options, base: Union[typing.Type[faust.app.base.AppCommand], NoneType] = None, **kwargs) → Callable[Callable, Type[faust.app.base.AppCommand]][source]¶

Return type:	`Callable`[[`Callable`], `Type`[`AppCommand`]]

FlowControlQueue(maxsize: int = None, *, clear_on_resume: bool = False, loop: asyncio.events.AbstractEventLoop = None) → mode.utils.queues.ThrowableQueue[source]¶

Like asyncio.Queue, but can be suspended/resumed.

Return type:	`ThrowableQueue`

Worker(**kwargs) → faust.app.base.WorkerT[source]¶

Return type:	`WorkerT`

on_webserver_init(web: faust.web.base.Web) → None[source]¶

Return type:	`None`

conf¶

producer[source]¶

consumer¶

transport¶: Message transport. :rtype: TransportT

tables[source]¶: Map of available tables, and the table manager service.

coroutine commit(self, topics: AbstractSet[Union[str, faust.types.tuples.TP]]) → bool[source]¶

Commit offset for acked messages in specified topics’.

Warning

This will commit acked messages in all topics if the topics argument is passed in as None.

Return type:	`bool`

logger = <Logger faust.app.base (WARNING)>¶

coroutine maybe_start_client(self) → None[source]¶

Start the app in Client-Only mode if not started as Server.

Return type:	`None`

maybe_start_producer[source]¶: Ensure producer is started.

coroutine on_stop(self) → None[source]¶

Called every time before the service is stopped/restarted.

Return type:	`None`

coroutine send(self, channel: Union[faust.types.channels.ChannelT, str], key: Union[bytes, faust.types.core.ModelT, typing.Any, NoneType] = None, value: Union[bytes, faust.types.core.ModelT, typing.Any] = None, partition: int = None, key_serializer: Union[faust.types.codecs.CodecT, str, NoneType] = None, value_serializer: Union[faust.types.codecs.CodecT, str, NoneType] = None, callback: Callable[faust.types.tuples.FutureMessage, Union[NoneType, typing.Awaitable[NoneType]]] = None) → Awaitable[faust.types.tuples.RecordMetadata][source]¶

Send event to channel/topic.

Parameters:

channel (Union[ChannelT[], str]) – Channel/topic or the name of a topic to send event to.
key (Union[bytes, ModelT, Any, None]) – Message key.
value (Union[bytes, ModelT, Any, None]) – Message value.
partition (Optional[int]) – Specific partition to send to. If not set the partition will be chosen by the partitioner.
key_serializer (Union[CodecT, str, None]) – Serializer to use (if value is not model).
value_serializer (Union[CodecT, str, None]) – Serializer to use (if value is not model).
callback (Optional[Callable[[FutureMessage[]], Union[None, Awaitable[None]]]]) –
Called after the message is fully delivered to the channel, but not to the consumer. Signature must be unary as the FutureMessage future is passed to it.

The resulting faust.types.tuples.RecordMetadata object is then available as fut.result().

Return type:

Awaitable[RecordMetadata]

coroutine start_client(self) → None[source]¶

Start the app in Client-Only mode necessary for RPC requests.

Notes

Once started as a client the app cannot be restarted as Server.

Return type:	`None`

topics[source]¶

Topic Conductor.

This is the mediator that moves messages fetched by the Consumer into the streams.

It’s also a set of registered topics by string topic name, so you can check if a topic is being consumed from by doing topic in app.topics.

monitor¶: Monitor keeps stats about what’s going on inside the worker. :rtype: Monitor[]

flow_control[source]¶

Internal flow control.

This object controls flow into stream queues, and can also clear all buffers.

http_client¶: HTTP Client Session. :rtype: HttpClientT

assignor[source]¶

Partition Assignor.

Responsible for partition assignment.

router[source]¶

Find the node partitioned data belongs to.

The router helps us route web requests to the wanted Faust node. If a topic is sharded by account_id, the router can send us to the Faust worker responsible for any account. Used by the @app.table_route decorator.

serializers[source]¶

label¶

shortlabel¶

`faust.app`¶

Faust

Navigation

Related Topics

Parameters:	silent (bool) – If true then import errors will be ignored. force (bool) – Force reading configuration immediately. By default the configuration will be read only when required.
Return type:	`None`

faust.app¶

`faust.app`¶