zabgrab.services.ports

Interfaces between the services and the outside world.

Services receive implementations of these protocols as arguments, so that tests can substitute fakes (a scripted UI, a frozen clock, an in-memory Zabbix server, …). Services never print, prompt or perform I/O except through these.

zabgrab.services.ports.ClientFactory

Creates an (unauthenticated) client for a server’s config.

Type:

type

Classes

Clock(*args, **kwargs)

Source of the current time.

Editor(*args, **kwargs)

Opens a file for the user to edit.

Option(label, value[, description])

One option offered by UI.choose() or UI.choose_many().

Progress(*args, **kwargs)

Reports progress of a download: an overall count plus one bar per item.

SecretStore(*args, **kwargs)

Persistent storage for API tokens, keyed by server URL.

UI(*args, **kwargs)

User interaction.

ZabbixClient(*args, **kwargs)

A connection to one Zabbix server.
class zabgrab.services.ports.Option(label, value, description=None)[source]

Bases: Generic[T]

One option offered by UI.choose() or UI.choose_many().

Parameters:

  • label (str)

  • value (T)

  • description (str | None)

label

What the user sees.

Type:

str

value

What choosing it returns.

Type:

T

description

Extra detail shown next to the label.

Type:

str or None

class zabgrab.services.ports.Progress(*args, **kwargs)[source]

Bases: Protocol

Reports progress of a download: an overall count plus one bar per item.

add_item(label, total)[source]

Start showing an item.

Parameters:

  • label (str) – What to call it.

  • total (int) – How many steps it has.

Returns:

int – An ID for the other methods.

Return type:

int

advance(item)[source]

Mark one step of an item (and of the overall total) as done.

Parameters:

item (int)

Return type:

None

remove_item(item)[source]

Stop showing an item.

Parameters:

item (int)

Return type:

None

class zabgrab.services.ports.UI(*args, **kwargs)[source]

Bases: Protocol

User interaction.

Messages are plain text; the implementation decides how to style them. They go to stderr, so stdout carries only data.

Every method that asks for input takes a fix argument: a message telling the user how to provide the answer without being asked (e.g. a flag or config snippet). When the session is not interactive, yes/no questions take their default answer with a warning; other questions raise zabgrab.errors.InputRequired carrying fix, since there’s no answer to carry on with.

property interactive: bool

Whether the user can be asked questions.

debug(message)[source]

Show a message only with -v or --debug.

Parameters:

message (str)

Return type:

None

info(message)[source]

Show a message (hidden by -q).

Parameters:

message (str)

Return type:

None

success(message)[source]

Show that something worked (hidden by -q).

Parameters:

message (str)

Return type:

None

warning(message)[source]

Show a warning (hidden by -qq).

Parameters:

message (str)

Return type:

None

error(message, *, details=(), hint=None)[source]

Show an error, with optional detail lines and a hint on how to fix it.

Parameters:
Return type:

None

confirm(question, *, fix, default=True)[source]

Ask a yes/no question.

Unlike the other questions, this doesn’t fail when the session isn’t interactive: it warns (including fix) and returns default.

Parameters:

  • question (str) – The question.

  • fix (str) – How to choose without being asked.

  • default (bool, optional) – The answer if the user just presses Enter, or can’t be asked.

Returns:

bool – The answer.

Return type:

bool

prompt(question, *, fix, parse, default=None, secret=False)[source]

Ask for free text.

Parameters:

  • question (str) – The question.

  • fix (str) – How to give the answer without being asked.

  • parse (callable) – Converts the answer, raising ValueError (whose message is shown before asking again) if it’s invalid.

  • default (str, optional) – The answer if the user just presses Enter.

  • secret (bool, optional) – Hide the input (e.g. for passwords).

Returns:

T – The parsed answer.

Raises:

InputRequired – If the session isn’t interactive.

Return type:

T

choose(question, options, *, fix, default=None)[source]

Ask the user to pick one option.

Parameters:

  • question (str) – The question.

  • options (sequence of Option) – The options.

  • fix (str) – How to choose without being asked.

  • default (int, optional) – The index of the option chosen if the user just presses Enter.

Returns:

T – The chosen option’s value.

Raises:

InputRequired – If the session isn’t interactive.

Return type:

T

choose_many(question, options, *, fix)[source]

Ask the user to pick one or more options (e.g. 1,3,5-8 or all).

Parameters:

  • question (str) – The question.

  • options (sequence of Option) – The options.

  • fix (str) – How to choose without being asked.

Returns:

list of T – The chosen options’ values, in the order offered.

Raises:

InputRequired – If the session isn’t interactive.

Return type:

list[T]

progress(description, total)[source]

Show progress of total steps while the context is active.

Parameters:

  • description (str)

  • total (int)

Return type:

AbstractContextManager[Progress]

class zabgrab.services.ports.Clock(*args, **kwargs)[source]

Bases: Protocol

Source of the current time.

now()[source]

Get the current time, timezone-aware.

Return type:

datetime

class zabgrab.services.ports.Editor(*args, **kwargs)[source]

Bases: Protocol

Opens a file for the user to edit.

edit(path)[source]

Open path and wait until the user is done.

Parameters:

path (Path)

Return type:

None

class zabgrab.services.ports.SecretStore(*args, **kwargs)[source]

Bases: Protocol

Persistent storage for API tokens, keyed by server URL.

get_token(url)[source]

Get the token stored for a server URL, if any.

Parameters:

url (str)

Return type:

str | None

set_token(url, token)[source]

Store a token for a server URL, replacing any stored before.

Parameters:
Return type:

None

delete_token(url)[source]

Delete a stored token; return whether there was one.

Parameters:

url (str)

Return type:

bool

urls()[source]

Get the URLs of the servers with stored tokens.

Return type:

list[str]

delete_all()[source]

Delete every stored token, leaving no trace; return how many there were.

Return type:

int

class zabgrab.services.ports.ZabbixClient(*args, **kwargs)[source]

Bases: Protocol

A connection to one Zabbix server.

See zabgrab.adapters.zabbix.ZabbixAPI for the real one.

async check_version()[source]

Get the server’s version, raising if it isn’t supported.

Return type:

str

async login_token(token)[source]

Authenticate with an API token, raising if it’s rejected.

Parameters:

token (str)

Return type:

None

async login_password(username, password)[source]

Authenticate with a username and password, raising if rejected.

Parameters:

  • username (str)

  • password (str)

Return type:

None

async close()[source]

Log out of any password session and close the connection.

Return type:

None

async hosts(*, pattern=None, group=None, limit=None)[source]

Find hosts by name pattern and host group pattern.

Parameters:

  • pattern (str | None)

  • group (str | None)

  • limit (int | None)

Return type:

list[Host]

async items(*, itemids=None, hostids=None, name=None, key=None, tag=None, limit=None)[source]

Find items matching all of the given filters.

Parameters:
Return type:

list[Item]

async history(item, start, end)[source]

Get raw values between two POSIX times (inclusive), oldest first.

Parameters:
Return type:

list[HistoryRow]

async trends(item, start, end)[source]

Get hourly trends for hours starting between two POSIX times (inclusive).

Parameters:
Return type:

list[TrendRow]