core.cli.core

Core Commands

Provides a framework for cli commands run against one or more onegov cloud applications.

OneGov cli commands are usually ran against an onegov.yml config file, which may contain definitions for multiple applications. It may define multiple application with different application classes, and it may contain wildcard applications which run the same application class, but contain multiple tenants for each application.

To have a command run against one or many applications we use a selector to help select the applications we want to target in a command.

In addition to selectors, onegov core cli commands provide a simple way to write a function that takes a request and an application. This function is then called for each application matching the selector, with a proper request and application context already setup (with the same characteristics as if called through an url in the browser).

Selector

A selector has the form <namespace>/<id>.

That is, it consists of the namespace of the application, and it’s id.

For example:

• /foo/bar

• /onegov_election_day/gr

• /onegov_town6/govikon

To select non-wildcard applications we can just omit the id:

• /foo

• /onegov_onboarding

Finally, to select multiple applications we can use wildcards:

• /foo/*

• /onegov_election_day/*

• /*/g??

Execution

To run a supported command we provide a selector as an option:

bin/onegov-core --select '/foo/*' subcommand

To find out what kind of selectors are available, we can simply run:

bin/onegov-core

Which will print out a list of selector suggestions.

Registering a Selector Based Command

To write a selector based command we first create a command group:

from onegov.core.cli import command_group
cli = command_group()

Using that command group, we can register our own commands:

@cli.command()
def my_click_command():
    pass

This command works like any other click command:

import click

@cli.command()
@click.option('--option')
def my_click_command(option):
    pass

Each command has the ability to influence the way selectors work. For example, a command which creates the path that matches the selector we can use:

@cli.command(context_settings={'creates_path': True})

By default, we expect that a selector is passed. For commands which usually run against all applications we can provide a default selector:

@cli.command(context_settings={'default_selector': '*'})

Using the app/request context

For a lot of commands the easiest approach is to have a function which is called for each application with a request. This allows us to write commands which behave like they were written in a view.

To do that we register a command which returns a function with the following signature:

def handle_command(request, app):
    pass

For example:

@cli.command()
def my_click_command():

    def handle_command(request, app):
        pass

    return handle_command

Setup like this, handle_command will be called with a request for each application (and tennant). This function acts exactly like a view. Most importantly, it does not require transaction commits, because like with ordinary requests, the transaction is automatically committed if no error occurs.

Using the app configurations directly

Sometimes we don’t want to use the request/app context, or maybe we want to setup something before receiving a request.

To do this, we use the pass_group_context decorator.

For example:

from onegov.core.cli import pass_group_context

@cli.command()
@pass_group_context
def my_click_command(group_context):

    for appcfg in group_context.appcfgs:
        # do something

This is independent of the app/request context. If we return a function, the function is going to be called with the request and the app. If we do not, the command ends as expected.

Returning multiple functions

When a cli command returns multiple functions, they are run in succession.

The signature is taken into account. If there’s a ‘request’ parameter in the function, the usual request context is set up.

If there is no ‘request’ parameter in the function, it is called once per appcfg, together with the group context:

@cli.command()
def my_special_command():

    def handle_command(request, app):
        pass

    def handle_raw(group_context, appcfg):
        pass

    return (handle_command, handle_raw)

Limiting Selectors to a Single Instance

Sometimes we want to write commands which only run against a single application. A good example is a command which returns 1/0 depending on the existence of something in an application.

To do that, we use:

@cli.command(context_settings={'singular': True})
def my_click_command():
    pass

If a selector is passed which matches more than one application, the command is not executed.

Module Contents

Classes

_GroupContextAttrs	Base class for protocol classes.
GroupContextGuard	Contains methods which abort the commandline program if any condition
GroupContext	Provides access to application configs for group commands.

Functions

get_context_specific_settings(→ ContextSpecificSettings)	Takes the given click context and extracts all context specific
run_processors(→ None)	Runs a sequence of processors either in a raw context or
command_group(→ click.Group)	Generates a click command group for individual modules.
abort(→ NoReturn)	Prints the given error message and aborts the program with a return

Attributes

CONTEXT_SPECIFIC_SETTINGS
pass_group_context

class core.cli.core._GroupContextAttrs

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing), for example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...

property available_selectors: list[str]

property matches: Iterator[str]

selector: str | None

matches_required: bool

singular: bool

creates_path: bool

core.cli.core.CONTEXT_SPECIFIC_SETTINGS = ('default_selector', 'creates_path', 'singular', 'matches_required')

class core.cli.core.GroupContextGuard

Bases: _GroupContextAttrs

Contains methods which abort the commandline program if any condition is not met.

Used as a mixin in GroupContext.

validate_guard_conditions(click_context: click.Context) → None

abort_if_no_selector(click_context: click.Context, matches: Collection[str]) → None

abort_if_no_subcommand(click_context: click.Context, matches: Collection[str]) → None

abort_if_no_match(click_context: click.Context, matches: Collection[str]) → None

abort_if_not_singular(click_context: click.Context, matches: Collection[str]) → None

abort_if_no_create_path(click_context: click.Context, matches: Collection[str]) → None

class core.cli.core.GroupContext(selector: str | None, config: dict[str, Any] | str | bytes, default_selector: str | None = None, creates_path: bool = False, singular: bool = False, matches_required: bool = True)

Bases: GroupContextGuard

Provides access to application configs for group commands.

Parameters:

• selector –

  Selects the applications which should be captured by a command_group().

  See Core Commands for more documentation about selectors.

• config – The targeted onegov.yml file or an equivalent dictionary.

• default_selector – The selector used if none is provided. If not given, a selector has to be provided.

• creates_path –

  True if the given selector doesn’t exist yet, but will be created. Commands which use this setting are expected to take a single path (no wildcards) and to create it during their runtime.

  Implies singular and matches_required.

• singular – True if the selector may not match multiple applications.

• matches_required – True if the selector must match at least one application.

property appcfgs: Iterator[ApplicationConfig]

Returns the matching appconfigs.

Since there’s only one appconfig per namespace, we ignore the path part of the selector and only focus on the namespace:

/namespace/application_id

property available_selectors: list[str]

Generates a list of available selectors.

The list doesn’t technically exhaust all options, but it returns all selectors targeting a single application as well as all selectors targeting a namespace by wildcard.

property all_wildcard_selectors: Iterator[str]

Returns all selectors targeting a namespace by wildcard.

property all_specific_selectors: Iterator[str]

Returns all selectors targeting an application directly.

property matches: Iterator[str]

Returns the specific selectors matching the context selector.

That is, a combination of namespace / application id is returned. Since we only know an exhaustive list of application id’s if we have a database connection this is currently limited to applications with one. Since we do not have any others yet that’s fine.

However if we implement a database-less application in the future which takes wildcard ids, we need some way to enumerate those ids.

See https://github.com/OneGov/onegov.core/issues/13

available_schemas(appcfg: onegov.server.config.ApplicationConfig) → list[str]

Returns all available schemas, if the application is database bound.

split_match(match: str) → tuple[str, str]

match_to_path(match: str) → str | None

Takes the given match and returns the application path used in http requests.

match_to_appcfg(match: str) → ApplicationConfig | None

Takes the given match and returns the maching appcfg object.

core.cli.core.get_context_specific_settings(context: click.Context) → ContextSpecificSettings

Takes the given click context and extracts all context specific settings from it.

core.cli.core.pass_group_context

core.cli.core.run_processors(group_context: GroupContext, processors: Sequence[Callable[..., Any]]) → None

Runs a sequence of processors either in a raw context or in a fully running application within a server.

This is extracted into its own utility function, so we can create commands that only require a server for the initial setup, but then may go on to run forever without the additional overhead (e.g. to implement a spooler)

core.cli.core.command_group() → click.Group

Generates a click command group for individual modules.

Each individual module may have its own command group from which to run commands to. Read https://click.pocoo.org/6/commands/ to learn more about command groups.

The returned command group will provide the individual commands with an optional list of applications to operate on and it allows commands to return a callback function which will be invoked with the app config (if available), an application instance and a request.

That is to say, the command group automates setting up a proper request context.

core.cli.core.abort(msg: str) → NoReturn

Prints the given error message and aborts the program with a return code of 1.