Skip to content

CLI#

FastStream has its own built-in CLI tool for your maximum comfort as a developer.

Thanks to typer and watchfiles. Their work is the basis of this tool.

Installation#

To install the FastStream CLI, you need to run the following command:

pip install faststream[cli]

After installation, you can check which commands are available by executing:

faststream --help


Usage: faststream [OPTIONS] COMMAND [ARGS]...

  Generate, run and manage FastStream apps to greater development experience

Options:
  -v, --version                   Show current platform, python and FastStream
                                  version
  --install-completion [bash|zsh|fish|powershell|pwsh]
                                  Install completion for the specified shell.
  --show-completion [bash|zsh|fish|powershell|pwsh]
                                  Show completion for the specified shell, to
                                  copy it or customize the installation.
  --help                          Show this message and exit.

Commands:
  docs  AsyncAPI schema commands
  run   Run [MODULE:APP] FastStream application

Running the Project#

Multiprocessing Scaling#

FastStream allows you to scale application right from the command line by running you application in the Process pool.

Just set the --worker option to scale your application:

faststream run serve:app --workers 2


INFO     - Started parent process [7591]
INFO     - Started child process [7593]
INFO     - Started child process [7594]
INFO     - test |            - `Handle` waiting for messages
INFO     - test |            - `Handle` waiting for messages

Hot Reload#

Thanks to watchfiles, written in Rust, you can work with your project easily. Edit the code as much as you like - the new version has already been launched and is waiting for your requests!

faststream run serve:app --reload

Tip

Please, install watchfiles if you want to use --reload feature

pip install watchfiles


INFO     - Started reloader process [7902] using WatchFiles
INFO     - FastStream app starting...
INFO     - test |            - `Handle` waiting for messages
INFO     - FastStream app started successfully! To exit press CTRL+C

By default FastStream watches for .py file changes, but you can specify an extra file extensions to watch by (your config files as an example)

faststream run serve:app --reload  --reload-ext .yml --realod-ext .yaml

Environment Management#

You can pass any custom flags and launch options to the FastStream CLI even without first registering them. Just use them when launching the application - and they will be right in your environment.

Use this option to select environment files, configure logging, or at your discretion.

For example, we will pass the .env file to the context of our application:

faststream run serve:app --env=.env.dev


INFO     - FastStream app starting...
INFO     - test |            - `Handle` waiting for messages
INFO     - FastStream app started successfully! To exit press CTRL+C


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream, ContextRepo
from faststream.kafka import KafkaBroker
from pydantic_settings import BaseSettings

broker = KafkaBroker()

app = FastStream(broker)

class Settings(BaseSettings):
    host: str = "localhost:9092"

@app.on_startup
async def setup(env: str, context: ContextRepo):
    settings = Settings(_env_file=env)
    await broker.connect(settings.host)
    context.set_global("settings", settings)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream, ContextRepo
from faststream.confluent import KafkaBroker
from pydantic_settings import BaseSettings

broker = KafkaBroker()

app = FastStream(broker)

class Settings(BaseSettings):
    host: str = "localhost:9092"

@app.on_startup
async def setup(env: str, context: ContextRepo):
    settings = Settings(_env_file=env)
    await broker.connect(settings.host)
    context.set_global("settings", settings)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream, ContextRepo
from faststream.rabbit import RabbitBroker
from pydantic_settings import BaseSettings

broker = RabbitBroker()

app = FastStream(broker)

class Settings(BaseSettings):
    host: str = "amqp://guest:guest@localhost:5672/" 

@app.on_startup
async def setup(env: str, context: ContextRepo):
    settings = Settings(_env_file=env)
    await broker.connect(settings.host)
    context.set_global("settings", settings)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream, ContextRepo
from faststream.nats import NatsBroker
from pydantic_settings import BaseSettings

broker = NatsBroker()

app = FastStream(broker)

class Settings(BaseSettings):
    host: str = "nats://localhost:4222"

@app.on_startup
async def setup(env: str, context: ContextRepo):
    settings = Settings(_env_file=env)
    await broker.connect(settings.host)
    context.set_global("settings", settings)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from faststream import FastStream, ContextRepo
from faststream.redis import RedisBroker
from pydantic_settings import BaseSettings

broker = RedisBroker()

app = FastStream(broker)

class Settings(BaseSettings):
    host: str = "redis://localhost:6379"

@app.on_startup
async def setup(env: str, context: ContextRepo):
    settings = Settings(_env_file=env)
    await broker.connect(settings.host)
    context.set_global("settings", settings)

Note

Note that the env parameter was passed to the setup function directly from the command line

All passed values can be of type bool, str or list[str].

In this case, the flags will be interpreted as follows:

faststream run app:app --flag             # flag = True
faststream run app:app --no-flag          # flag = False
faststream run app:app --my-flag          # my_flag = True
faststream run app:app --key value        # key = "value"
faststream run app:app --key 1 2          # key = ["1", "2"]
faststream run app:app --key 1 --key 2    # key = ["1", "2"]

You can use them both individually and together in unlimited quantities.

AsyncAPI Schema#

Also, the FastStream CLI allows you to work with the AsyncAPI schema in a simple way.

You are able to generate .json or .yaml files by your application code or host HTML representation directly:

faststream docs --help


Usage: faststream docs [OPTIONS] COMMAND [ARGS]...

  AsyncAPI schema commands

Options:
  --help  Show this message and exit.

Commands:
  gen    Generate project AsyncAPI schema
  serve  Serve project AsyncAPI schema

To learn more about the commands above, please visit AsyncAPI export and AsyncAPI hosting.