Faust

https://faust-streaming.github.io/faust
Admin
Faust is a stream processing library for Python that ports ideas from Kafka Streams, enabling the...
Tokens:430,987
Snippets:3,698
Trust Score:8.3
Update:2 months ago
Show doc for...
Tokens:
Raw
### Setup development environment

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Commands to install dependencies and configure the development environment.

```console
$ make develop
```

```console
$ make hooks
```

```console
$ make cdevelop
```

--------------------------------

### Install the codec extension

Source: https://faust-streaming.github.io/faust/userguide/models.html

Install the package locally using setup.py.

```bash
$ python setup.py install
```

--------------------------------

### Start web server

Source: https://faust-streaming.github.io/faust/_modules/faust/web/drivers/aiohttp.html

Initializes the runner and starts the web site.

```python
async def start_server(self) -> None:
    """Start the web server."""
    await self._runner.setup()
    site = self._create_site()
    await site.start()
```

--------------------------------

### Running the stock ordering system

Source: https://faust-streaming.github.io/faust/_sources/userguide/livecheck.rst.txt

Starts the worker instance for the stock ordering system example.

```console
$ python examples/livecheck.py worker -l info
```

--------------------------------

### Install a custom codec extension

Source: https://faust-streaming.github.io/faust/_modules/faust/serializers/codecs.html

Install the newly created codec extension using setup.py.

```console
$ python setup.py install
```

--------------------------------

### Start a worker for the agent example

Source: https://faust-streaming.github.io/faust/_sources/userguide/agents.rst.txt

Run the worker for the defined agent module.

```console
$ faust -A examples.agent worker -l info
```

--------------------------------

### Install Faust from source

Source: https://faust-streaming.github.io/faust/userguide/installation.html

Manual installation steps for building Faust from a downloaded source archive.

```bash
$ tar xvfz faust-0.0.0.tar.gz
$ cd faust-0.0.0
$ python setup.py build
# python setup.py install
```

--------------------------------

### Install project dependencies

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Install the required dependencies for development.

```console
$ pip install -U -r requirements/dist.txt
```

--------------------------------

### Install and Run Project Commands

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Use development mode to install the package and execute the registered console script.

```console
$ python setup.py develop
```

```console
$ proj worker -l info
```

--------------------------------

### Start Faust worker via command line

Source: https://faust-streaming.github.io/faust/_modules/faust/types/settings/settings.html

Example command to start a Faust worker specifying the application module.

```console
faust -A example.simple worker
```

--------------------------------

### Install multiple extensions

Source: https://faust-streaming.github.io/faust/_sources/introduction.rst.txt

Combine multiple extension bundles into a single installation command using commas.

```bash
$ pip install faust-streaming[uvloop,fast,rocksdb,datadog,redis]
```

--------------------------------

### Install optimization bundles

Source: https://faust-streaming.github.io/faust/_sources/introduction.rst.txt

Install a collection of performance-related packages using the fast bundle.

```bash
pip install faust-streaming[fast]
```

```bash
pip install faust-streaming[aiodns]
```

```bash
pip install faust-streaming[cchardet]
```

```bash
pip install faust-streaming[ciso8601]
```

```bash
pip install faust-straming[cython]
```

```bash
pip install faust-streaming[orjson]
```

```bash
pip install faust-streaming[setproctitle]
```

--------------------------------

### Install Faust development version

Source: https://faust-streaming.github.io/faust/userguide/installation.html

Install the latest snapshot directly from the GitHub repository.

```bash
$ pip install https://github.com/faust-streaming/faust/zipball/master#egg=faust
```

--------------------------------

### Start service conditionally

Source: https://faust-streaming.github.io/faust/_modules/mode/services.html

Checks the service state and initiates startup only if it has not already been started.

```python
async def maybe_start(self) -> bool:
        """Start the service, if it has not already been started."""
        if not self._started.is_set():
            await self.start()
            return True
        return False
```

--------------------------------

### app.main()

Source: https://faust-streaming.github.io/faust/userguide/application.html

Starts the Faust command-line program using arguments from sys.argv.

```APIDOC
## app.main()

### Description
Starts the Faust command-line program. This method allows the script to support standard Faust CLI arguments and subcommands.
```

--------------------------------

### Start Kafka Services

Source: https://faust-streaming.github.io/faust/_sources/playbooks/leaderelection.rst.txt

Commands to initialize Zookeeper and Kafka servers.

```console
$ $KAFKA_HOME/bin/zookeeper-server-start $KAFKA_HOME/etc/kafka/zookeeper.properties
```

```console
$ $KAFKA_HOME/bin/kafka-server-start $KAFKA_HOME/etc/kafka/server.properties
```

--------------------------------

### Install test dependencies and run tests

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Commands to install required test dependencies and execute the test suite.

```console
$ pip install -U -r requirements/test.txt
$ pip install -U -r requirements/default.txt
```

```console
$ py.test
```

```console
$ py.test t/unit/test_app.py
```

--------------------------------

### Install Faust with feature bundles

Source: https://faust-streaming.github.io/faust/userguide/installation.html

Install Faust along with specific optional dependencies using bracket notation.

```bash
$ pip install "faust-streaming[rocksdb]"
```

```bash
$ pip install "faust-streaming[rocksdb,uvloop,fast,redis]"
```

--------------------------------

### Start app directly

Source: https://faust-streaming.github.io/faust/userguide/application.html

Start an app instance directly without the worker's signal handlers or logging configuration.

```python
async def start_app(app):
    await app.start()
```

```python
def start_in_loop(app):
    loop = asyncio.get_event_loop()
    loop.ensure_future(app.start())
```

--------------------------------

### on_start()

Source: https://faust-streaming.github.io/faust/_modules/faust/livecheck/app.html

Lifecycle method called when the LiveCheck application starts.

```APIDOC
## on_start()

### Description
Initializes the LiveCheck application, including installing the bus agent and test execution agent.
```

--------------------------------

### agents()

Source: https://faust-streaming.github.io/faust/reference/faust.app.base.html

Returns a list of services required to start agents.

```APIDOC
## agents()

### Description
Returns a list of services required to start agents.

### Returns
- **Iterable[ServiceT]** - A list of services.
```

--------------------------------

### Configure setup.py for Project Packaging

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Define project metadata and dependencies in setup.py to enable installation.

```python
#!/usr/bin/env python
from setuptools import find_packages, setup

setup(
    name='proj',
    version='1.0.0',
    description='Use Faust to blah blah blah',
    author='Ola Normann',
    author_email='ola.normann@example.com',
    url='http://proj.example.com',
    platforms=['any'],
    license='Proprietary',
    packages=find_packages(exclude=['tests', 'tests.*']),
    include_package_data=True,
    zip_safe=False,
    install_requires=['faust'],
    python_requires='~=3.10',
)
```

--------------------------------

### Instantiate and Configure App in Steps

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Configure the application instance after initialization, similar to the Flask factory pattern.

```pycon
>>> app = faust.App('myApp')
>>> app.conf.broker='kafka://kafka.example.com'
>>> app.conf.store='rocksdb://'
>>> app.main()
```

--------------------------------

### Define setup.py for extension

Source: https://faust-streaming.github.io/faust/_sources/userguide/models.rst.txt

Basic setup.py file for packaging a Faust codec extension.

```python
import setuptools
```

--------------------------------

### Define an Agent for Testing

Source: https://faust-streaming.github.io/faust/userguide/testing.html

Example setup of a Faust application, record, topic, table, and agent to be used in subsequent tests.

```python
app = faust.App('test-example')

class Order(faust.Record, serializer='json'):
    account_id: str
    product_id: str
    amount: int
    price: float

orders_topic = app.topic('orders', value_type=Order)
orders_for_account = app.Table('order-count-by-account', default=int)

@app.agent(orders_topic)
async def order(orders):
    async for order in orders.group_by(Order.account_id):
        orders_for_account[order.account_id] += 1
        yield order
```

--------------------------------

### Run a stream processor as an asyncio task

Source: https://faust-streaming.github.io/faust/userguide/streams.html

A practical example of running a stream processor as an app task, ensuring the worker is started.

```python
import faust

class Withdrawal(faust.Record):
    account: str
    amount: float

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

withdrawals_topic = app.topic('withdrawals', value_type=Withdrawal)

@app.task
async def mytask():
    async for w in withdrawals_topic.stream():
        print(w.amount)

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

--------------------------------

### Define Setup for Codec Extension

Source: https://faust-streaming.github.io/faust/_modules/faust/serializers/codecs.html

Configure a setuptools entry point to register a codec extension package.

```python
from setuptools import setup, find_packages

setup(
    name='faust-msgpack',
    version='1.0.0',
    description='Faust msgpack serialization support',
    author='Ola A. Normann',
    author_email='ola@normann.no',
    url='http://github.com/example/faust-msgpack',
    platforms=['any'],
    license='BSD',
    packages=find_packages(exclude=['ez_setup', 'tests', 'tests.*']),
    zip_safe=False,
    install_requires=['msgpack-python'],
    tests_require=[],
    entry_points={
        'faust.codecs': [
            'msgpack = faust_msgpack:msgpack',
        ],
    },
)
```

--------------------------------

### Start Web Server

Source: https://faust-streaming.github.io/faust/_modules/faust/web/drivers/aiohttp.html

Initializes the server and manages the service dependency, either in a thread or as a standard service.

```python
async def on_start(self) -> None:
    """Call when the embedded web server starts.

    Only used for `faust worker`, not when using :meth:`wsgi`.
    """
    cors = self.cors
    assert cors
    self.init_server()
    server_cls = ServerThread if self.app.conf.web_in_thread else Server
    self._thread = server_cls(self, loop=self.loop, beacon=self.beacon)
    self.add_dependency(self._thread)
```

--------------------------------

### Start AIOKafka Producer

Source: https://faust-streaming.github.io/faust/_modules/faust/transport/drivers/aiokafka.html

Initializes and starts the AIOKafka producer instance.

```python
async def on_start(self) -> None:
    """Call when producer starts."""
    await super().on_start()
    producer = self._producer = self._new_producer()
    self.beacon.add(producer)
    await producer.start()
```

--------------------------------

### maybe_start_client

Source: https://faust-streaming.github.io/faust/_modules/faust/app/base.html

Starts the application in Client-Only mode only if it has not already been started as a server.

```APIDOC
## maybe_start_client()

### Description
Checks the application state and starts it in client-only mode if it is not currently running.
```

--------------------------------

### Service start lifecycle methods

Source: https://faust-streaming.github.io/faust/_modules/mode/services.html

Methods to initiate service startup and default startup procedures.

```python
    async def start(self) -> None:
        await self._default_start()


    async def _default_start(self) -> None:
        loop = self.loop
        assert loop  # make sure loop is set
        assert not self._started.is_set()
        self._started.set()
        await self._actually_start()
```

--------------------------------

### enumerate(start)

Source: https://faust-streaming.github.io/faust/reference/faust.html

Enumerate values received on this stream, working with async generators.

```APIDOC
## enumerate(start=0)

### Description
Enumerate values received on this stream. Unlike Python’s built-in enumerate, this works with async generators.

### Parameters
- **start** (int) - Optional - The starting index.

### Returns
- **AsyncIterable[Tuple[int, T_co]]** - An async iterable of tuples containing the index and the value.
```

--------------------------------

### Configure Faust Autodiscovery

Source: https://faust-streaming.github.io/faust/reference/faust.html

Examples of configuring the autodiscover parameter for the Faust App instance.

```python
app = App(..., autodiscover=get_all_packages_to_scan)
```

```python
app = App(..., autodiscover=True)
```

```python
app = App(..., autodiscover=['package1', 'package2'])
```

```python
app = App(.., autodiscover=False)
```

--------------------------------

### Add Console Script Entry Point

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Register a command-line interface for the application within setup.py.

```python
setup(
    ...,
    entry_points={
        'console_scripts': [
            'proj = proj.app:main',
        ],
    },
)
```

--------------------------------

### Install OpenTelemetry instrumentation

Source: https://faust-streaming.github.io/faust/_sources/userguide/asgi.rst.txt

Install the necessary dependencies for OpenTelemetry support in Faust.

```console
$ pip install "faust-streaming[opentelemetry]"
```

--------------------------------

### Install pytest-asyncio

Source: https://faust-streaming.github.io/faust/_sources/userguide/testing.rst.txt

Install the required extension for running asynchronous tests with pytest.

```console
$ pip install -U pytest-asyncio
```

--------------------------------

### Configure Setuptools for Faust Projects

Source: https://faust-streaming.github.io/faust/userguide/application.html

Using setup.py to define project metadata and console scripts for easier execution.

```python
#!/usr/bin/env python
from setuptools import find_packages, setup

setup(
    name='proj',
    version='1.0.0',
    description='Use Faust to blah blah blah',
    author='Ola Normann',
    author_email='ola.normann@example.com',
    url='http://proj.example.com',
    platforms=['any'],
    license='Proprietary',
    packages=find_packages(exclude=['tests', 'tests.*']),
    include_package_data=True,
    zip_safe=False,
    install_requires=['faust'],
    python_requires='~=3.10',
)
```

```python
setup(
    ...,
    entry_points={
        'console_scripts': [
            'proj = proj.app:main',
        ],
    },
)
```

```bash
$ python setup.py develop
```

```bash
$ proj worker -l info
```

--------------------------------

### app.http_client.get

Source: https://faust-streaming.github.io/faust/reference/faust.html

Performs an asynchronous HTTP GET request using the Faust application's internal HTTP client.

```APIDOC
## GET app.http_client.get

### Description
Performs an asynchronous HTTP GET request. This is recommended for network operations within Faust agents.

### Parameters
- **url** (str) - Required - The URL to request.
- **timeout** (float) - Optional - Timeout in seconds for the request.
```

--------------------------------

### Install debugging extras

Source: https://faust-streaming.github.io/faust/_sources/introduction.rst.txt

Install tools for monitoring and debugging Faust applications.

```bash
pip install faust-streaming[debug]
```

```bash
pip install faust-streaming[aiomonitor]
```

```bash
pip install faust-streaming[setproctitle]
```

--------------------------------

### Install Fickling dependency

Source: https://faust-streaming.github.io/faust/userguide/models.html

Install the required extra package to use the pickle_fickling codec.

```bash
$ pip install "faust-streaming[fickling]"
```

--------------------------------

### Start Faust Application with app.main

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Initializes the application as a command-line program using sys.argv.

```python
# examples/command.py
import faust

app = faust.App('umbrella-command-example')

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

```console
$ python examples/command.py
```

```console
$ python examples/command.py worker --help
```

--------------------------------

### Install Fickling extra

Source: https://faust-streaming.github.io/faust/_sources/userguide/models.rst.txt

Install the optional Fickling dependency for enhanced pickle analysis.

```console
$ pip install "faust-streaming[fickling]"
```

--------------------------------

### Initialize Dependencies

Source: https://faust-streaming.github.io/faust/_modules/faust/app/base.html

Returns a list of additional service dependencies to be started with the application.

```python
def on_init_dependencies(self) -> Iterable[ServiceT]:
    """Return list of additional service dependencies.

    The services returned will be started with the
    app when the app starts.
    """
    # Add the main Monitor sensor.
    # The beacon is also reattached in case the monitor
    # was created by the user.
    self.monitor.beacon.reattach(self.beacon)
    self.monitor.loop = self.loop
    self.sensors.add(self.monitor)

    if self.producer_only:
        return self.boot_strategy.producer_only()
    elif self.client_only:
        return self.boot_strategy.client_only()
    else:
        return self.boot_strategy.server()
```

--------------------------------

### Installing OpenTelemetry Support

Source: https://faust-streaming.github.io/faust/_sources/userguide/sensors.rst.txt

Install the necessary dependencies to enable OpenTelemetry metrics reporting.

```console
$ pip install "faust[opentelemetry]"
```

--------------------------------

### Install Faust ASGI dependencies

Source: https://faust-streaming.github.io/faust/_sources/userguide/asgi.rst.txt

Command to install the necessary dependencies for FastAPI integration.

```console
$ pip install "faust-streaming[fastapi]"
```

--------------------------------

### Install aiomonitor

Source: https://faust-streaming.github.io/faust/_sources/userguide/debugging.rst.txt

Install the aiomonitor library directly or as part of the Faust debug bundle.

```console
$ pip install aiomonitor
```

```console
$ pip install -U faust-streaming[debug]
```

--------------------------------

### Param Initialization and Configuration

Source: https://faust-streaming.github.io/faust/_modules/faust/types/settings/params.html

The constructor defines the metadata and behavior for a setting, including versioning, default values, and related CLI options.

```python
    def __init__(
        self,
        *,
        name: str,
        env_name: Optional[str] = None,
        default: Optional[IT] = None,
        default_alias: Optional[str] = None,
        default_template: Optional[str] = None,
        allow_none: Optional[bool] = None,
        ignore_default: Optional[bool] = None,
        section: Optional[_Section] = None,
        version_introduced: Optional[str] = None,
        version_deprecated: Optional[str] = None,
        version_removed: Optional[str] = None,
        version_changed: Optional[Mapping[str, str]] = None,
        deprecation_reason: Optional[str] = None,
        related_cli_options: Optional[Mapping[str, List[str]]] = None,
        related_settings: Optional[List[Any]] = None,
        help: Optional[str] = None,
        **kwargs: Any,
    ) -> None:
        assert name
        self.name = name
        self.storage_name = f"_{name}"
        if env_name is not None:
            self.env_name = env_name
        if default is not None:
            self.default = default
        if default_alias is not None:
            self.default_alias = default_alias
        if default_template is not None:
            self.default_template = default_template
        if allow_none is not None:
            self.allow_none = allow_none
        if ignore_default is not None:
            self.ignore_default = ignore_default
        if section is not None:
            self.section = section
        assert self.section
        if version_introduced is not None:
            self.version_introduced = version_introduced
        if version_deprecated is not None:
            self.version_deprecated = version_deprecated
        if version_removed is not None:
            self.version_removed = version_removed
        if version_changed is not None:
            self.version_changed = version_changed
        if deprecation_reason is not None:
            self.deprecation_reason = deprecation_reason
        if help is not None:
            self.__doc__ = help
        self._on_get_value_: Optional[OutputCallable] = None
        self._on_set_default_: Optional[OnDefaultCallable] = None
        self.options = kwargs
        self.related_cli_options = related_cli_options or {}
        self.related_settings = related_settings or []
        self._init_options(**self.options)

        if self.version_deprecated:
            assert self.deprecation_reason
```

--------------------------------

### Build documentation

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Commands to install documentation dependencies and generate the HTML documentation.

```console
$ pip install -U -r requirements/docs.txt
```

```console
$ cd docs
$ rm -rf _build
$ make html
```

--------------------------------

### Render README file

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Install distribution requirements and run the make command to update the project README.

```console
$ pip install -U requirements/dist.txt
$ make readme
```

--------------------------------

### Install Faust via pip

Source: https://faust-streaming.github.io/faust/userguide/installation.html

Standard command to install the latest version of the faust-streaming package.

```bash
$ pip install -U faust-streaming
```

--------------------------------

### Configure setuptools for a Faust plugin

Source: https://faust-streaming.github.io/faust/_sources/userguide/models.rst.txt

Use the entry_points section to register the codec with Faust's plugin system.

```python
setuptools.setup(
    name='faust-msgpack',
    version='1.0.0',
    description='Faust msgpack serialization support',
    author='Ola A. Normann',
    author_email='ola@normann.no',
    url='http://github.com/example/faust-msgpack',
    platforms=['any'],
    license='BSD',
    packages=find_packages(exclude=['ez_setup', 'tests', 'tests.*']),
    zip_safe=False,
    install_requires=['msgpack-python'],
    tests_require=[],
    entry_points={
        'faust.codecs': [
            'msgpack = faust_msgpack:msgpack',
        ],
    },
)
```

--------------------------------

### Initialize Settings

Source: https://faust-streaming.github.io/faust/_modules/faust/types/settings/base.html

Configures settings by iterating through defined parameters, checking environment variables, and applying default values.

```python
    def _init_settings(self, kwargs: Dict) -> None:
        # called by Settings.__init__(**kwargs) to configure
        # available settings.
        conf = cast(_Settings, self)
        for name, param in self.SETTINGS.items():
            if not param.ignore_default:
                provided_value = kwargs.get(name, None)
                if param.env_name is not None:
                    provided_env_value = self.getenv(param.env_name)
                    if provided_env_value is not None:
                        provided_value = kwargs[name] = provided_env_value
                param.on_init_set_value(conf, provided_value)
        for name, param in self.SETTINGS.items():
            if not param.ignore_default:
                provided_value = kwargs.get(name, None)
                param.on_init_set_default(conf, provided_value)
```

--------------------------------

### Install Faust with Eventlet support

Source: https://faust-streaming.github.io/faust/faq.html

Install the required bundle to enable eventlet integration with Faust.

```bash
$ pip install -U faust-streaming[eventlet]
```

--------------------------------

### Implement on_start lifecycle method

Source: https://faust-streaming.github.io/faust/_modules/faust/tables/manager.html

Defines the startup sequence for the table manager, including recovery initialization.

```python
async def on_start(self) -> None:
    """Call when table manager is starting."""
    await self.sleep(1.0)
    if not self.should_stop:
        await self._update_channels()
        await self.recovery.start()
```

--------------------------------

### Start worker from synchronous code

Source: https://faust-streaming.github.io/faust/userguide/application.html

Use execute_from_commandline to start a worker instance from a synchronous context.

```python
>>> worker = Worker(app)
>>> worker.execute_from_commandline()
```

--------------------------------

### Start LiveCheck components

Source: https://faust-streaming.github.io/faust/userguide/livecheck.html

Commands to initialize the stock ordering system and the LiveCheck instance in separate terminal sessions.

```bash
$ python examples/livecheck.py worker -l info
```

```bash
$ python examples/livecheck.py livecheck -l info
```

--------------------------------

### Install RocksDB on MacOS Sierra

Source: https://faust-streaming.github.io/faust/_sources/introduction.rst.txt

Requires specific compiler flags to install python-rocksdb on MacOS Sierra.

```bash
$ CFLAGS='-std=c++11 -stdlib=libc++ -mmacosx-version-min=10.10' \
    pip install -U --no-cache python-rocksdb
```

--------------------------------

### Configure Application via Attributes

Source: https://faust-streaming.github.io/faust/userguide/application.html

Instantiate an application and configure its settings via the conf attribute before starting.

```python
>>> app = faust.App('myApp')
>>> app.conf.broker='kafka://kafka.example.com'
>>> app.conf.store='rocksdb://'
>>> app.main()
```

--------------------------------

### Execute service startup logic

Source: https://faust-streaming.github.io/faust/_modules/mode/services.html

Handles the internal startup sequence, including dependencies, exit stacks, and task initialization.

```python
    async def _actually_start(self) -> None:  # noqa: C901
        """Start the service."""
        # NOTE: ``should_stop`` is re-read into a local before every check:
        # the flag flips while the awaits below are suspended, and mypy
        # otherwise folds repeated reads of the property into the result of
        # the first one and reports the later checks as dead code.
        for _ in [1]:  # to use break
            if not self.restart_count:
                for dep in self.on_init_dependencies():
                    self.add_dependency(dep)
                await self.on_first_start()
                should_stop = self.should_stop
                if should_stop:
                    break
            self.exit_stack.__enter__()
            await self.async_exit_stack.__aenter__()
            should_stop = self.should_stop
            if should_stop:
                break
            try:
                self._log_mundane("Starting...")
                await self.on_start()
                should_stop = self.should_stop
                if should_stop:
                    break
                for task in self._get_tasks():
                    self.add_future(task.fun(self))
                for child in self._children:
                    if child is not None:
                        await child.maybe_start()
                    should_stop = self.should_stop
                    if should_stop:
                        break
                self.log.debug("Started.")
                await self.on_started()
            except BaseException:
                self.exit_stack.__exit__(*sys.exc_info())
                await self.async_exit_stack.__aexit__(*sys.exc_info())
                raise
```

--------------------------------

### Install individual Faust extensions

Source: https://faust-streaming.github.io/faust/_sources/introduction.rst.txt

Use these commands to install specific optional dependencies for Faust Streaming.

```bash
pip install faust-streaming[rocksdb]
```

```bash
pip install faust-streaming[redis]
```

```bash
pip install faust-streaming[datadog]
```

```bash
pip install faust-streaming[statsd]
```

```bash
pip install faust-streaming[uvloop]
```

```bash
pip install faust-streaming[eventlet]
```

```bash
pip install faust-streaming[yaml]
```

--------------------------------

### Register extensions in setup.py

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Add the new feature name to the EXTENSIONS set in setup.py.

```python
EXTENSIONS = {
    'debug',
    'fast',
    'rocksdb',
    'uvloop',
}
```

--------------------------------

### Install click_completion Dependency

Source: https://faust-streaming.github.io/faust/_sources/history/changelog-1.1.rst.txt

Commands to verify and install the optional click_completion dependency for shell completion features.

```console
$ ./examples/withdrawals.py completion
Usage: withdrawals.py completion [OPTIONS]

Error: Missing required dependency, but this is easy to fix.
Run `pip install click_completion` from your virtualenv
and try again!
```

```console
$ pip install click_completion
[...]
```

--------------------------------

### Start Agent Instance

Source: https://faust-streaming.github.io/faust/_modules/faust/agents/agent.html

Initiates a single agent instance, managing concurrency and partition assignment.

```python
    async def _start_one(
        self,
        *,
        index: Optional[int] = None,
        active_partitions: Optional[Set[TP]] = None,
        stream: Optional[StreamT] = None,
        channel: Optional[ChannelT] = None,
    ) -> ActorT:
        # an index of None means there's only one instance,
        # and `index is None` is used as a test by functions that
        # disallows concurrency.
        index = index if self.concurrency > 1 else None
        return await self._start_task(
            index=index,
            active_partitions=active_partitions,
            stream=stream,
            channel=channel,
            beacon=self.beacon,
        )
```

--------------------------------

### Initialize Settings Class

Source: https://faust-streaming.github.io/faust/_modules/faust/types/settings/settings.html

The Settings class constructor accepts various configuration parameters for brokers, agents, consumers, and producers.

```python
class Settings(base.SettingsRegistry):
    NODE_HOSTNAME: ClassVar[str] = socket.gethostname()
    DEFAULT_BROKER_URL: ClassVar[str] = "kafka://localhost:9092"

    _id: str
    _name: str
    _version: int

    #: Environment.
    #: Defaults to :data:`os.environ`.
    env: Mapping[str, str]

    def __init__(
        self,
        id: str,
        *,
        # Common settings:
        autodiscover: Optional[AutodiscoverArg] = None,
        datadir: Optional[typing.Union[str, Path]] = None,
        tabledir: Optional[typing.Union[str, Path]] = None,
        debug: Optional[bool] = None,
        env_prefix: Optional[str] = None,
        id_format: Optional[str] = None,
        origin: Optional[str] = None,
        timezone: Optional[typing.Union[str, tzinfo]] = None,
        version: Optional[int] = None,
        # Agent settings:
        agent_supervisor: Optional[SymbolArg[Type[SupervisorStrategyT]]] = None,
        # Broker settings:
        broker: Optional[BrokerArg] = None,
        broker_consumer: Optional[BrokerArg] = None,
        broker_producer: Optional[BrokerArg] = None,
        broker_api_version: Optional[str] = None,
        broker_check_crcs: Optional[bool] = None,
        broker_client_id: Optional[str] = None,
        broker_commit_every: Optional[int] = None,
        broker_commit_interval: Optional[Seconds] = None,
        broker_commit_livelock_soft_timeout: Optional[Seconds] = None,
        broker_credentials: Optional[CredentialsArg] = None,
        broker_heartbeat_interval: Optional[Seconds] = None,
        broker_max_poll_interval: Optional[Seconds] = None,
        broker_max_poll_records: Optional[int] = None,
        broker_rebalance_timeout: Optional[Seconds] = None,
        broker_request_timeout: Optional[Seconds] = None,
        broker_session_timeout: Optional[Seconds] = None,
        ssl_context: Optional[ssl.SSLContext] = None,
        # Consumer settings:
        consumer_api_version: Optional[str] = None,
        consumer_max_fetch_size: Optional[int] = None,
        consumer_auto_offset_reset: Optional[str] = None,
        consumer_group_instance_id: Optional[str] = None,
        consumer_metadata_max_age_ms: Optional[int] = None,
        consumer_connections_max_idle_ms: Optional[int] = None,
        # Topic serialization settings:
        key_serializer: Optional[CodecArg] = None,
        value_serializer: Optional[CodecArg] = None,
        # Logging settings:
        logging_config: Optional[Mapping] = None,
        loghandlers: Optional[List[logging.Handler]] = None,
        # Producer settings:
        producer_acks: Optional[int] = None,
        producer_api_version: Optional[str] = None,
        producer_compression_type: Optional[str] = None,
        producer_linger_ms: Optional[int] = None,
        producer_max_batch_size: Optional[int] = None,
        producer_max_request_size: Optional[int] = None,
        producer_partitioner: Optional[SymbolArg[PartitionerT]] = None,
        producer_request_timeout: Optional[Seconds] = None,
        producer_threaded: bool = False,
        producer_metadata_max_age_ms: Optional[int] = None,
        producer_connections_max_idle_ms: Optional[int] = None,
        # RPC settings:
```

--------------------------------

### Start a Faust worker

Source: https://faust-streaming.github.io/faust/userguide/application.html

Run the worker command to start consuming messages from the agent's topic.

```bash
$ python examples/agent.py worker -l info
```

--------------------------------

### Define agent service initialization

Source: https://faust-streaming.github.io/faust/_modules/faust/app/base.html

Returns the list of services required to start agents.

```python
    def agents(self) -> Iterable[ServiceT]:
        """Return list of services required to start agents."""
        return [self.app.agents]
```

--------------------------------

### Start ASGI service

Source: https://faust-streaming.github.io/faust/_modules/faust/contrib/asgi.html

Lifecycle method to initialize and start the ASGI server, including OpenTelemetry instrumentation.

```python
    async def on_start(self) -> None:
        """Start serving."""
        if self.asgi_app is None:
            raise ImproperlyConfigured("AsgiService requires an ASGI application")
        maybe_instrument_opentelemetry(self.asgi_app, self.opentelemetry)
        self._server = self._create_server()
        self._serve_fut = self.add_future(self._server.serve())
```

--------------------------------

### Create a Table

Source: https://faust-streaming.github.io/faust/_sources/userguide/tables.rst.txt

Initialize a new table using the application instance.

```python
table = app.Table('totals', default=int)
```

--------------------------------

### Start Faust in client-only mode

Source: https://faust-streaming.github.io/faust/userguide/application.html

Use this method to initialize the application for RPC communication without starting the worker.

```python
await app.start_client()
```

--------------------------------

### Start worker via command line

Source: https://faust-streaming.github.io/faust/userguide/application.html

Commands to start a worker instance from the terminal using the -A flag.

```bash
$ faust -A myproj worker -l info
```

```bash
$ faust -A myproj:faust_app worker -l info
```

--------------------------------

### Verify configuration documentation

Source: https://faust-streaming.github.io/faust/_sources/contributing.rst.txt

Ensure all settings are documented in the configuration reference.

```console
$ make configcheck
```

--------------------------------

### Start app in client-only mode

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Initialize the app for RPC requests without starting a full worker process.

```python
await app.start_client()
```

--------------------------------

### Install Free-Threaded Environment

Source: https://faust-streaming.github.io/faust/_sources/developerguide/free_threading.rst.txt

Commands to set up a Python 3.14t environment using uv for free-threading development.

```console
$ uv python install 3.14t
$ uv venv --python 3.14t
$ uv pip install -r requirements/freethreading.txt 'Cython>=3.1' setuptools setuptools_scm
```

--------------------------------

### Initialize Settings Component

Source: https://faust-streaming.github.io/faust/_modules/faust/types/settings/settings.html

Initializes the settings object with an ID and version, and sets up the environment prefix.

```python
    def on_init(self, id: str, **kwargs: Any) -> None:
        # version is required for the id
        # and the id is required as a component in several default
        # setting values so we hack this in here to make sure
        # it's set.
        self._init_env_prefix(**kwargs)
        self._version = kwargs.get("version", 1)
        self.id = id
```

--------------------------------

### Configure Application Parameters

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Set the broker URL and storage driver during application initialization.

```pycon
>>> app = faust.App(
...     'myid',
...     broker='kafka://kafka.example.com',
...     store='rocksdb://',
... )
```

--------------------------------

### Start Faust app in background

Source: https://faust-streaming.github.io/faust/_sources/userguide/application.rst.txt

Schedule the app to start in the current event loop to allow other tasks to run concurrently.

```python
def start_in_loop(app):
    loop = asyncio.get_event_loop()
    loop.ensure_future(app.start())
```

--------------------------------

### app.main()

Source: https://faust-streaming.github.io/faust/_modules/faust/app/base.html

Executes the faust umbrella command using the current application instance.

```APIDOC
## main()

### Description
Executes the faust umbrella command using this app. This method initializes the worker and starts the CLI.

### Signature
`main() -> NoReturn`
```

--------------------------------

### Start a Faust worker with an existing event loop

Source: https://faust-streaming.github.io/faust/_modules/faust/worker.html

Demonstrates how to manually initialize a worker and start it within an existing asyncio event loop.

```python
async def start_worker(worker: Worker) -> None:
    await worker.start()

def manage_loop():
    loop = asyncio.new_event_loop()
    worker = Worker(app, loop=loop)
    loop.run_until_complete(start_worker(worker))
```
Tokens:
Raw