faust.cli.base

Command-line programs using https://pypi.org/project/click/.

class faust.cli.base.AppCommand(ctx: Context, *args: Any, key_serializer: Optional[Union[CodecT, str]] = None, value_serializer: Optional[Union[CodecT, str]] = None, **kwargs: Any)

Command that takes -A app as argument.

abstract: ClassVar[bool] = False

require_app = True

value_serialier: Optional[Union[CodecT, str]]

The codec used to serialize values. Taken from instance parameters or value_serializer.

classmethod from_handler(*options: Any, **kwargs: Any) → Callable[[Callable], Type[AppCommand]]

Decorate async def command to create command class.

Return type:

_CallableGenericAlias[_CallableType, _GenericAlias[AppCommand]]

app: AppT

key_serializer: Optional[Union[CodecT, str]]

The codec used to serialize keys. Taken from instance parameters or key_serializer.

async on_stop() → None

Call after command executed.

Return type:

None

to_key(typ: Optional[str], key: str) → Any

Convert command-line argument string to model (key).

Parameters:

• typ (_UnionGenericAlias[str, None]) – The name of the model to create.

• key (str) – The string json of the data to populate it with.

Notes

Uses key_serializer to set the codec for the key (e.g. "json"), as set by the --key-serializer option.

Return type:

Any

to_value(typ: Optional[str], value: str) → Any

Convert command-line argument string to model (value).

Parameters:

• typ (_UnionGenericAlias[str, None]) – The name of the model to create.

• key – The string json of the data to populate it with.

Notes

Uses value_serializer to set the codec for the value (e.g. "json"), as set by the --value-serializer option.

Return type:

Any

to_model(typ: Optional[str], value: str, serializer: Optional[Union[CodecT, str]]) → Any

Convert command-line argument to model.

Generic version of to_key()/to_value().

Parameters:

• typ (_UnionGenericAlias[str, None]) – The name of the model to create.

• key – The string json of the data to populate it with.

• serializer (_UnionGenericAlias[CodecT, str, None]) – The argument setting it apart from to_key/to_value enables you to specify a custom serializer not mandated by key_serializer, and value_serializer.

Notes

Uses value_serializer to set the codec for the value (e.g. "json"), as set by the --value-serializer option.

Return type:

Any

import_relative_to_app(attr: str) → Any

Import string like “module.Model”, or “Model” to model class.

Return type:

Any

to_topic(entity: str) → Any

Convert topic name given on command-line to app.topic().

Return type:

Any

abbreviate_fqdn(name: str, *, prefix: str = '') → str

Abbreviate fully-qualified Python name, by removing origin.

app.conf.origin is the package where the app is defined, so if this is examples.simple it returns the truncated:

>>> app.conf.origin
'examples.simple'
>>> abbr_fqdn(app.conf.origin,
...           'examples.simple.Withdrawal',
...           prefix='[...]')
'[...]Withdrawal'

but if the package is not part of origin it provides the full path:

>>> abbr_fqdn(app.conf.origin,
...           'examples.other.Foo', prefix='[...]')
'examples.other.foo'

Return type:

str

property blocking_timeout: float

Return the blocking timeout used for this command. :rtype: float

class faust.cli.base.Command(ctx: Context, *args: Any, **kwargs: Any)

Base class for subcommands.

exception UsageError(message: str, ctx: Optional[Context] = None)

An internal exception that signals a usage error. This typically aborts any further handling.

Parameters:

• message – the error message to display.

• ctx – optionally the context that caused this error. Click will fill in the context automatically in some situations.

exit_code = 2

The exit code for this exception.

show(file: Optional[IO] = None) → None

Return type:

None

abstract: ClassVar[bool] = True

daemon: bool = False

redirect_stdouts: Optional[bool] = None

redirect_stdouts_level: Optional[int] = None

builtin_options: Sequence[Callable[[Any], Any]] = [<function option.<locals>.decorator>, option('--app', '-A', help='Path of Faust application to use, or the name of a module.'), option('--quiet/--no-quiet', '-q', default=False, help='Silence output to <stdout>/<stderr>.'), option('--debug/--no-debug', default=False, help='Enable debugging output, and the blocking detector.'), option('--workdir', '-W', default=None, type=<click.types.Path object>, help='Working directory to change to after start.'), option('--datadir', '-D', default='{conf.name}-data', type=<click.types.Path object>, help='Directory to keep application state.'), option('--json', default=False, is_flag=True, help='Return output in machine-readable JSON format'), option('--loop', '-L', default='aio', type=Choice(['aio', 'eventlet', 'uvloop']), help='Event loop implementation to use.'), option('--logfile', '-f', callback=<function compat_option.<locals>._callback>, expose_value=False, default=None, type=<click.types.Path object>, help='Path to logfile (default is <stderr>).'), option('--loglevel', '-l', callback=<function compat_option.<locals>._callback>, expose_value=False, default='WARN', type=Choice(['crit', 'error', 'warn', 'info', 'debug']), help='Logging level to use.'), option('--blocking-timeout', callback=<function compat_option.<locals>._callback>, expose_value=False, default=None, type=<class 'float'>, help='when --debug: Blocking detector timeout.'), option('--console-port', callback=<function compat_option.<locals>._callback>, expose_value=False, default=50101, type=<TCPPort 1<=x<=65535>, help='when --debug: Port to run debugger console on.')]

options: Optional[MutableSequence[Callable[[Any], Any]]] = None

classmethod as_click_command() → Callable

Convert command into https://pypi.org/project/click/ command.

Return type:

_CallableType

classmethod parse(argv: Sequence[str]) → Mapping

Parse command-line arguments in argv and return mapping.

Return type:

_SpecialGenericAlias

debug: bool

quiet: bool

workdir: str

datadir: str

json: bool

logfile: str

stdout: IO

stderr: IO

args: Tuple

kwargs: Dict

prog_name: str = ''

async run(*args: Any, **kwargs: Any) → Any

Override this method to define what your command does.

async execute(*args: Any, **kwargs: Any) → Any

Execute command.

Return type:

Any

async on_stop() → None

Call after command executed.

Return type:

None

run_using_worker(*args: Any, **kwargs: Any) → NoReturn

Execute command using faust.Worker.

Return type:

_SpecialForm

on_worker_created(worker: Worker) → None

Call when creating faust.Worker to execute this command.

Return type:

None

as_service(loop: AbstractEventLoop, *args: Any, **kwargs: Any) → ServiceT

Wrap command in a mode.Service object.

Return type:

ServiceT

worker_for_service(service: ServiceT, loop: Optional[AbstractEventLoop] = None) → Worker

Create faust.Worker instance for this command.

Return type:

Worker

tabulate(data: Sequence[Sequence[str]], headers: Optional[Sequence[str]] = None, wrap_last_row: bool = True, title: str = '', **kwargs: Any) → str

Create an ANSI representation of a table of two-row tuples.

See also

Keyword arguments are forwarded to terminaltables.SingleTable

Note

If the --json option is enabled this returns json instead.

Return type:

str

table(data: Sequence[Sequence[str]], title: str = '', **kwargs: Any) → BaseTable

Format table data as ANSI/ASCII table.

Return type:

BaseTable

say(message: str, file: Optional[IO] = None, err: Optional[IO] = None, **kwargs: Any) → None

Print something to stdout (or use file=stderr kwarg).

Note

Does not do anything if the --quiet option is enabled.

Return type:

None

carp(s: Any, **kwargs: Any) → None

Print something to stdout (or use file=stderr kwargs).

Note

Does not do anything if the --debug option is enabled.

Return type:

None

dumps(obj: Any) → str

Serialize object using JSON.

Return type:

str

property loglevel: str

Return the log level used for this command. :rtype: str

property blocking_timeout: float

Return the blocking timeout used for this command. :rtype: float

property console_port: int

Return the https://pypi.org/project/aiomonitor/ console port. :rtype: int

class faust.cli.base.argument(*args: Any, **kwargs: Any)

Create command-line argument.

SeeAlso:

click.argument()

faust.cli.base.find_app(app: str, *, symbol_by_name: ~typing.Callable = <function symbol_by_name>, imp: ~typing.Callable = <function import_from_cwd>, attr_name: str = 'app') → AppT

Find app by string like examples.simple.

Notes

This function uses import_from_cwd to temporarily add the current working directory to PYTHONPATH, such that when importing the app it will search the current working directory last.

You can think of it as temporarily running with the PYTHONPATH set like this:

You can disable this with the imp keyword argument, for example passing imp=importlib.import_module.

Examples

>>> # If providing the name of a module, it will attempt
>>> # to find an attribute name (.app) in that module.
>>> # Example below is the same as importing::
>>> #    from examples.simple import app
>>> find_app('examples.simple')

>>> # If you want an attribute other than .app you can
>>> # use : to separate module and attribute.
>>> # Examples below is the same as importing::
>>> #     from examples.simple import my_app
>>> find_app('examples.simple:my_app')

>>> # You can also use period for the module/attribute separator
>>> find_app('examples.simple.my_app')

Return type:

AppT

class faust.cli.base.option(*args: Any, show_default: bool = True, **kwargs: Any)

Create command-line option.

SeeAlso:

click.option()