precept package

class precept.Argument(*flags: str, type: Optional[type] = None, help: Optional[str] = None, choices: Optional[Iterable] = None, default: Optional[Any] = None, nargs: Optional[Union[str, int]] = None, action: Optional[str] = None, required: Optional[bool] = None, metavar: Optional[str] = None, dest: Optional[str] = None)[source]

Bases: precept._immutable.ImmutableDict

Argument of a Command, can either be optional or not depending on the flags

__init__(*flags: str, type: Optional[type] = None, help: Optional[str] = None, choices: Optional[Iterable] = None, default: Optional[Any] = None, nargs: Optional[Union[str, int]] = None, action: Optional[str] = None, required: Optional[bool] = None, metavar: Optional[str] = None, dest: Optional[str] = None)[source]
Parameters
  • flags – How to call the argument, prefixing with - makes the argument a keyword. Example: '-d', '--date' make the variable available as date, but can be supplied as -d from the cli.

  • type – The type of the variable to cast to, default to str.

  • help – Description to go along with

  • choices – The available choices to choose from.

  • default – The value to take if not supplied.

  • nargs – Number of times the argument can be supplied.

  • action – What to do with the argument.

  • required – Makes a keyword argument required.

  • metavar – Name in help.

  • dest – The name of the variable to add the value to once parsed.

action
choices
default
dest
property flag_key
flags
help
metavar
nargs
register(parser)[source]

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

required
type
class precept.AsyncExecutor(loop=None, executor=None, max_workers=None)[source]

Bases: object

Execute functions in a Pool Executor

__init__(loop=None, executor=None, max_workers=None)[source]
Parameters
  • loop – asyncio event loop.

  • executor – Set to use an already existing PoolExecutor, default to a new ThreadPoolExecutor if not supplied.

  • max_workers – Max workers of the created ThreadPoolExecutor.

async execute(func, *args, **kwargs)[source]

Execute a sync function asynchronously in the executor.

Parameters
  • func – Synchronous function.

  • args – Argument to give to the function.

  • kwargs – Keyword arguments to give to the function

Returns

async execute_with_lock(func, *args, **kwargs)[source]

Acquire lock before executing the function.

Parameters
  • func – Synchronous function.

  • args

  • kwargs

Returns

wraps(func)[source]

Wraps a synchronous function to execute in the pool when called, making it async.

Parameters

func – The function to wraps

Returns

Async wrapped function.

class precept.AutoNameEnum(value)[source]

Bases: enum.Enum

An enumeration.

class precept.Command(*arguments: precept._cli.Argument, name: Optional[str] = None, description: Optional[str] = None, help: Optional[str] = None, auto: bool = False, services: Optional[List[precept._services.Service]] = None)[source]

Bases: object

Command decorator, methods of CliApp subclasses decorated with this gets a sub-command in the parser.

Wrapped methods will gets the arguments by the Argument flag.

__init__(*arguments: precept._cli.Argument, name: Optional[str] = None, description: Optional[str] = None, help: Optional[str] = None, auto: bool = False, services: Optional[List[precept._services.Service]] = None)[source]
arguments: Iterable[precept._cli.Argument]
property command_name
description: str
register(subparsers)[source]
class precept.Config(config_format: precept._configs.ConfigFormat = ConfigFormat.TOML, root_name='config')[source]

Bases: precept._configs.Nestable

Root config class, assign ConfigProperties as class members.

__init__(config_format: precept._configs.ConfigFormat = ConfigFormat.TOML, root_name='config')[source]
property config_format: precept._configs.ConfigFormat
read_dict(data: dict)[source]
read_file(path: str)[source]
save(path: str)[source]
class precept.ConfigFormat(value)[source]

Bases: precept._tools.AutoNameEnum

Available formats to use with configs.

  • TOML provided by tomlkit, supports comments and more complex types.

  • YML provided by ruamel.yaml, supports comments and more complex types.

  • JSON stdlib, no support for comments and types.

  • INI stdlib, support for comments.

INI = 'ini'
JSON = 'json'
TOML = 'toml'
YML = 'yml'
serializer(config)[source]
class precept.ConfigProperty(default=None, comment=None, config_type=None, environ_name=None, auto_environ=False, name=None, auto_global=False, global_name=None)[source]

Bases: object

__init__(default=None, comment=None, config_type=None, environ_name=None, auto_environ=False, name=None, auto_global=False, global_name=None)[source]
class precept.ImmutableDict(**kwargs)[source]

Bases: collections.abc.Mapping

__init__(**kwargs)[source]
class precept.ImmutableMeta(name, bases, attributes)[source]

Bases: abc.ABCMeta

class precept.ImmutableProp[source]

Bases: object

class precept.Nestable(parent=None, parent_len=0)[source]

Bases: collections.abc.Mapping

__init__(parent=None, parent_len=0)[source]
get_prop_paths(parent='')[source]
get_root(current=None)[source]
class precept.Plugin[source]

Bases: object

Plugin’s are automatically added to a precept application upon installation.

Set the entry point in setup to register the plugin:

entry_point = {‘{app_name}.plugins’: [‘plugin = my_plugin:plugin’]}

name: str = ''
async setup(application)[source]

Setup the plugin

Parameters

application (precept.Precept) – The running precept application.

Returns

class precept.Precept(config_file: Optional[Union[str, List[str]]] = None, loop=None, executor=None, executor_max_workers=None, add_dump_config_command=False, help_formatter=<class 'precept._cli.CombinedFormatter'>, logger_level=20, logger_fmt=None, logger_datefmt=None, logger_stream=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, logger_colors=None, logger_style='%', services: Optional[List[precept._services.Service]] = None, print_version: bool = True)[source]

Bases: object

Auto cli generator, methods decorated with Command will have a corresponding sub-command in the cli application.

Commands will get the arguments named as the last element of the command flags.

Override main method for root handler, it gets all the global_arguments

__init__(config_file: Optional[Union[str, List[str]]] = None, loop=None, executor=None, executor_max_workers=None, add_dump_config_command=False, help_formatter=<class 'precept._cli.CombinedFormatter'>, logger_level=20, logger_fmt=None, logger_datefmt=None, logger_stream=<_io.TextIOWrapper name='<stderr>' mode='w' encoding='UTF-8'>, logger_colors=None, logger_style='%', services: Optional[List[precept._services.Service]] = None, print_version: bool = True)[source]
Parameters
  • config_file – Path to the default config file to use. Can be specified with --config-file

  • loop – Asyncio loop to use.

  • executor – concurrent executor to use.

  • add_dump_config_command – Add a dump-config command.

  • help_formatter – The cli formatter to use.

  • logger_level – Set logger level when setting up logging.

  • logger_fmt – The format of the logger.

  • logger_datefmt – Date format of the logger.

  • logger_stream – The stream to print the logs.

  • logger_colors – Dictionary with key logger level name and values of bg/fg/style dict.

  • logger_style – The symbol to use for formatting.

  • services – List of global services to start with the program.

  • print_version – Print the version & name of the app before start.

config: precept._configs.Config = None
config_class = None
property config_path
default_configs: dict = {}
global_arguments = []
async main(**kwargs)[source]

Handler when no command has been entered. Gets the globals arguments.

Parameters

kwargs – Global arguments.

Returns

prog_name = ''
async setup_plugins()[source]

Load and setup the registered plugins.

To register a plugin, subclass Plugin and instantiate then add to setup.py entry_points:

‘{app_name}.plugins’: [‘my_plugin = plugin_module:plugin’]

Returns

async setup_services(command: Optional[precept._cli.Command] = None)[source]

Setup the services for the command or the main application.

Parameters

command – The command that was run.

Returns

start(args=None)[source]

Start the application loop.

Returns

async start_services(command: Optional[precept._cli.Command] = None)[source]

Start the services, automatically called by start.

If the application if run with another method you can call this to start the global services without the command argument.

Parameters

command – The command that was run.

Returns

async stop_services(command: Optional[precept._cli.Command] = None)[source]

Stop the services, automatically called by start.

Call this if your application is not run with start and you have running services.

Parameters

command – The command that was run.

Returns

version = '0.0.1'
class precept.Service(events: Optional[precept.events._dispatcher.EventDispatcher] = None)[source]

Bases: object

Service’s runs alongside the main application.

Communicate via events.

Events
  • {name}_setup when added to services.

  • {name}_start after calling start.

  • {name}_stop after calling stop.

__init__(events: Optional[precept.events._dispatcher.EventDispatcher] = None)[source]
name: str = 'service'
async setup(application)[source]

Called when added to services.

Use this to set events and other post initialization that may need the application instance.

Parameters

application (precept.Precept) – Precept application

Returns

async start()[source]

Start the service.

Returns

async stop()[source]

Stop the service.

Returns

precept.config_factory(data, root=None, key=None)[source]
precept.is_windows()[source]

Errors

exception precept.errors.ConfigError[source]

Bases: precept.errors.PreceptError

Error in the config system.

exception precept.errors.ImmutableError[source]

Bases: precept.errors.PreceptError

Immutable properties cannot change

exception precept.errors.PreceptError[source]

Bases: Exception

Base exception thrown by precept

Subpackages