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

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

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

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

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

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

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

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

Return type:

_CallableType

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

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[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:

_SpecialForm

on_worker_created(worker: Worker) None[source]

Call when creating faust.Worker to execute this command.

Return type:

None

as_service(loop: AbstractEventLoop, *args: Any, **kwargs: Any) ServiceT[source]

Wrap command in a mode.Service object.

Return type:

ServiceT

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

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

Format table data as ANSI/ASCII table.

Return type:

BaseTable

say(message: str, file: Optional[IO] = None, err: Optional[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: 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)[source]

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[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.option(*args: Any, show_default: bool = True, **kwargs: Any)[source]

Create command-line option.

SeeAlso:

click.option()