faust-streaming

Faust is organized into several core modules that handle different aspects of stream processing:

• faust.app: Defines the Faust application, including configuration and message production.
• faust.cli: The command-line interface.
• faust.exceptions: Custom exceptions.
• faust.models: Defines how message keys and values are serialized/deserialized.
• faust.sensors: Records statistics from a running application.
• faust.serializers: Handles JSON serialization and encoding codecs.
• faust.stores: Manages table storage (e.g., in-memory, RocksDB).
• faust.streams: Implementation of Streams and Tables.
• faust.topics: Tools for creating and managing topic descriptions.
• faust.transport: Message transport implementations (e.g., aiokafka).
• faust.types: Public interface for static typing.
• faust.utils: General utilities.
• faust.web: Web abstractions and the Faust web server.
• faust.windows: Windowing strategies.
• faust.worker: Deployment helpers for signal handling and graceful shutdown.

LiveCheck is an end-to-end testing tool designed for production or staging environments in microservice architectures or any asynchronous system (e.g., a monolith sending Celery tasks).

It acts as a passive observer that tracks requests as they travel through your system and allows you to define contracts that must be met at every step. Because it is a passive observer, it can detect and alert when subsystems are down or when anomalies occur (e.g., checking if an account debt exceeds a threshold after a specific change).

Key features:

• Probabilistic Execution: Tests are executed based on a defined probability (e.g., 0.1%, 30%, or 50% of requests), allowing you to run tests on a subset of production traffic.
• Stream Processor Integration: Every LiveCheck test case is a stream processor, meaning it can utilize Faust tables to store and manage stateful data during the test lifecycle.

An agent is a distributed system processing events in a stream. Each event is a key/value pair, typically described using faust.Record for type safety and serialization.

To create an agent, use the @app.agent decorator on an async function. The function must take a stream as its argument and iterate over it using async for.

Partitioning Behavior:

• By Key: Messages with the same key are always delivered to the same agent instance. This is useful for stateful processing.
• Round-Robin: If no key is set (key=None), messages are distributed evenly across available workers. This is ideal for distributing work in a cluster.

Fault Tolerance: If a worker fails, Kafka moves the partition to an online worker. Faust uses "standby tables" and a custom partition manager to optimize startup time and availability.

import faust

app = faust.App('example', broker='kafka://localhost:9092')

@app.agent()
async def myagent(stream):
    async for event in stream:
        ...  # process event

A Service is a component that can be started and stopped. Faust is composed of many services, including App, Stream, Agent, Table, and TableManager.

Services are built using the mode.Service class. You can register a service class with your app using the @app.service decorator. Services can implement on_start and on_stop lifecycle methods and can run background tasks using the @Service.task decorator.

import faust
from mode import Service

app = faust.App('service-example')

@app.service
class MyService(Service):

    async def on_start(self):
        print('MYSERVICE IS STARTING')

    async def on_stop(self):
        print('MYSERVICE IS STOPPING')

    @Service.task
    async def _background_task(self):
        while not self.should_stop:
            print('BACKGROUND TASK WAKE UP')
            await self.sleep(1.0)

if __name__ == '__main__':
    app.main()

Faust uses a layered abstraction model to decouple message processing from the underlying transport mechanism. This allows agents to work with different types of communication backends.

• Agent: The processing unit that iterates over streams.
• Stream: An abstraction that iterates over channels.
• Channel: A buffer or queue used to send and receive messages. Channels can be local (in-memory) or network-based.
• Topic: A named channel backed by a transport (like a Kafka topic).
• Transport: The underlying driver (e.g., aiokafka) that handles the actual network communication.

Abstraction Hierarchy:

• For local processing: Agent <--> Stream <--> Channel
• For distributed processing: Agent <--> Stream <--> Topic <--> Transport <--> aiokafka

Because Topics are highly Kafka-specific, you should subclass Channels if you need to implement different communication means like RabbitMQ, MQTT, or ZeroMQ.

Faust supports a Blueprint pattern for defining reusable web components. A web.Blueprint can contain routes and views, which are then registered to a Faust app with an optional url_prefix.

from faust import web

blueprint = web.Blueprint('user')

@blueprint.route('/', name='list')
class UserListView(web.View):
    async def get(self, request: web.Request) -> web.Response:
        return self.json({'hello': 'world'})

@blueprint.route('/{username}/', name='detail')
class UserDetailView(web.View):
    async def get(self, request: web.Request) -> web.Response:
        name = request.match_info['username']
        return self.json({'hello': name})

    async def post(self, request: web.Request) -> web.Response:
        ...

    async def delete(self, request: web.Request) -> web.Response:
        ...

# Register the blueprint to an app
blueprint.register(app, url_prefix='/users/')