faust.cli.base

Command-line programs using click.

faust.cli.base.argument(*args, **kwargs) → Callable[Any, Any][source]

Create command-line argument.

SeeAlso:
click.argument()
Return type:Callable[[Any], Any]
faust.cli.base.option(*option_decls, show_default: bool = True, **kwargs) → Callable[Any, Any][source]

Create command-line option.

SeeAlso:
func:click.option
Return type:Callable[[Any], Any]
faust.cli.base.find_app(app: str, *, symbol_by_name: Callable = <function symbol_by_name>, imp: Callable = <function import_from_cwd>) → 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, **kwargs) → 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>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>, <function option.<locals>.decorator>]
options = None
classmethod as_click_command() → Callable[source]
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 = ''
run_using_worker(*args, **kwargs) → NoReturn[source]
Return type:_NoReturn
on_worker_created(worker: mode.worker.Worker) → None[source]
Return type:None
as_service(loop: asyncio.events.AbstractEventLoop, *args, **kwargs) → mode.services.Service[source]
Return type:Service[]
worker_for_service(service: mode.types.services.ServiceT, loop: asyncio.events.AbstractEventLoop = None) → mode.worker.Worker[source]
Return type:Worker[]
tabulate(data: Sequence[Sequence[str]], headers: Sequence[str] = None, wrap_last_row: bool = True, title: str = '', title_color: str = 'blue', **kwargs) → 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) → 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) → 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) → 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]
Return type:str
loglevel
Return type:str
blocking_timeout
Return type:float
console_port
Return type:int
coroutine execute(self, *args, **kwargs) → Any[source]
Return type:Any
coroutine on_stop(self) → None[source]
Return type:None
coroutine run(self, *args, **kwargs) → Any[source]
class faust.cli.base.AppCommand(ctx: click.core.Context, *args, key_serializer: Union[faust.types.codecs.CodecT, str, None] = None, value_serializer: Union[faust.types.codecs.CodecT, str, None] = None, **kwargs) → None[source]

Command that takes -A app as argument.

abstract = False
require_app = True
coroutine on_stop(self) → None[source]
Return type:None
value_serialier = None

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

classmethod from_handler(*options, **kwargs) → Callable[Callable, Type[faust.cli.base.AppCommand]][source]
Return type:Callable[[Callable], Type[AppCommand]]
key_serializer = None

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

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