faust.cli.base

Command-line programs using click.

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

Create command-line argument.

SeeAlso:

click.argument()

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

Create command-line option.

SeeAlso:

click.option()

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

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.Command(ctx: click.core.Context, *args: Any, **kwargs: Any) → None[source]

Base class for subcommands.

exception UsageError(message, ctx=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
show(file=None)
abstract = True
daemon = False
redirect_stdouts = None
redirect_stdouts_level = None
builtin_options = [<function version_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('--no-color/--color', '--no_color/--color', default=False, help='Enable colors in output.'), 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=10.0, type=<class 'float'>, help='when --debug: Blocking detector timeout.'), option('--console-port', callback=<function compat_option.<locals>._callback>, expose_value=False, default=50101, type=IntRange(1, 65535), help='when --debug: Port to run debugger console on.')]
options = None
classmethod as_click_command() → Callable[source]

Convert command into click command.

Return type

Callable

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

Parse command-line arguments in argv and return mapping.

Return type

Mapping[~KT, +VT_co]

prog_name = ''
async run(*args: Any, **kwargs: Any) → Any[source]

Override this method to define what your command does.

async execute(*args: Any, **kwargs: Any) → Any[source]

Execute command.

Return type

Any

async on_stop() → None[source]

Call after command executed.

Return type

None

run_using_worker(*args: Any, **kwargs: Any) → NoReturn[source]

Execute command using faust.Worker.

Return type

_NoReturn

on_worker_created(worker: mode.worker.Worker) → None[source]

Call when creating faust.Worker to execute this command.

Return type

None

as_service(loop: asyncio.events.AbstractEventLoop, *args: Any, **kwargs: Any) → mode.services.Service[source]

Wrap command in a mode.Service object.

Return type

Service[]

worker_for_service(service: mode.types.services.ServiceT, loop: asyncio.events.AbstractEventLoop = None) → mode.worker.Worker[source]

Create faust.Worker instance for this command.

Return type

Worker[]

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

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) → terminaltables.base_table.BaseTable[source]

Format table data as ANSI/ASCII table.

Return type

BaseTable

color(name: str, text: str) → str[source]

Return text having a certain color by name.

Examples::
>>> self.color('blue', 'text_to_color')
>>> self.color('hiblue', text_to_color')

See also

colorclass: for a list of available colors.

Return type

str

dark(text: str) → str[source]

Return cursor text.

Return type

str

bold(text: str) → str[source]

Return text in bold.

Return type

str

bold_tail(text: str, *, sep: str = '.') → str[source]

Put bold emphasis on the last part of a foo.bar.baz string.

Return type

str

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

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[source]

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[source]

Serialize object using JSON.

Return type

str

property loglevel

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

property blocking_timeout

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

property console_port

Return the aiomonitor console port. :rtype: int

class faust.cli.base.AppCommand(ctx: click.core.Context, *args: Any, key_serializer: Union[faust.types.codecs.CodecT, str, None] = None, value_serializer: Union[faust.types.codecs.CodecT, str, None] = None, **kwargs: Any) → None[source]

Command that takes -A app as argument.

abstract = False
require_app = True
value_serialier = None

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

classmethod from_handler(*options: Any, **kwargs: Any) → Callable[Callable, Type[faust.cli.base.AppCommand]][source]

Decorate async def command to create command class.

Return type

Callable[[Callable], Type[AppCommand]]

key_serializer = None

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

async on_stop() → None[source]

Call after command executed.

Return type

None

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

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

Parameters

  • typ (Optional[str]) – 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[source]

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

Parameters

  • typ (Optional[str]) – 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: Union[faust.types.codecs.CodecT, str, None]) → Any[source]

Convert command-line argument to model.

Generic version of to_key()/to_value().

Parameters

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

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

  • serializer (Union[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[source]

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

Return type

Any

to_topic(entity: str) → Any[source]

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

Return type

Any

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

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