API Reference¶

The pyinfra API is designed to be used as follows:

Create the state we are going to operate on, which consists of:
- An inventory pyinfra.api.Inventory containing hosts pyinfra.api.Host, plus any data
- A config pyinfra.api.Config for global flag
- A state pyinfra.api.State that combines the inventory & config
Now that state is setup, we define operations:
- pyinfra.api.operation.add_op
- pyinfra.api.add_deploy
Now that's done, we execute it:
- pyinfra.api.operations.run_ops

Currently the best example of this in action is in pyinfra's own main.py.

Core API¶

pyinfra.api ¶

arguments ¶

all_global_arguments ¶

all_global_arguments() -> list[tuple[str, type]]

Return all global arguments and their types.

pop_global_arguments ¶

pop_global_arguments(
    state: State, host: Host, kwargs: dict[str, Any]
) -> tuple[AllArguments, list[str]]

Pop and return operation global keyword arguments, in preferred order:

From the current context (a direct @operator or @deploy function being called)
From any current @deploy context (deploy kwargs)
From the host data variables
From the config variables

command ¶

make_formatted_string_command ¶

make_formatted_string_command(
    string: str, *args, **kwargs
) -> StringCommand

Helper function that takes a shell command or script as a string, splits it using shlex.split and then formats each bit, returning a StringCommand instance with each bit.

Useful to enable string formatted commands/scripts, for example:

.. code:: python

curl_command = make_formatted_string_command(
    'curl -sSLf {0} -o {1}',
    QuoteString(src),
    QuoteString(dest),
)

config ¶

Config ¶

Config(**kwargs)

Bases: ConfigDefaults

The default/base configuration options for a pyinfra deploy.

connect ¶

connect_all ¶

connect_all(state: State)

Connect to all the configured servers in parallel. Reads/writes state.inventory.

Parameters:

Name	Type	Description	Default
`state`	``pyinfra.api.State`` obj	the state containing an inventory to connect to	required

disconnect_all ¶

disconnect_all(state: State)

Disconnect from all of the configured servers in parallel. Reads/writes state.inventory.

Parameters:

Name	Type	Description	Default
`state`	``pyinfra.api.State`` obj	the state containing an inventory to connect to	required

deploy ¶

Deploys come in two forms: on-disk, eg deploy.py, and @deploy wrapped functions. The latter enable re-usable (across CLI and API based execution) pyinfra extension creation (eg pyinfra-openstack).

add_deploy ¶

add_deploy(
    state: State,
    deploy_func: Callable[..., Any],
    *args,
    **kwargs,
) -> None

Prepare & add an deploy to pyinfra.state by executing it on all hosts.

Parameters:

Name	Type	Description	Default
`state`	``pyinfra.api.State`` obj	the deploy state to add the operation	required
`deploy_func`	`function`	the operation function from one of the modules,	required
`args/kwargs`		passed to the operation function	required

deploy ¶

deploy(
    name: str | None = None,
    data_defaults: dict | None = None,
) -> Callable[[Callable[P, Any]], PyinfraOperation[P]]

Decorator that takes a deploy function (normally from a pyinfra_* package) and wraps any operations called inside with any deploy-wide kwargs/data.

exceptions ¶

PyinfraError ¶

Bases: Exception

Generic pyinfra exception.

ConnectError ¶

Bases: PyinfraError

Exception raised when connecting fails.

FactError ¶

Bases: PyinfraError

Exception raised during fact gathering staging if a fact is unable to generate output/change state.

FactTypeError ¶

Bases: FactError, TypeError

Exception raised when a fact is passed invalid argument types.

FactValueError ¶

Bases: FactError, ValueError

Exception raised when a fact is passed invalid argument values.

FactProcessError ¶

Bases: FactError, RuntimeError

Exception raised when the data gathered for a fact cannot be processed.

OperationError ¶

Bases: PyinfraError

Exception raised during fact gathering staging if an operation is unable to generate output/change state.

OperationTypeError ¶

Bases: OperationError, TypeError

Exception raised when an operation is passed invalid argument types.

OperationValueError ¶

Bases: OperationError, ValueError

Exception raised when an operation is passed invalid argument values.

NestedOperationError ¶

Bases: OperationError

Exception raised when a nested (immediately executed) operation fails.

FactNotCollected ¶

Bases: FactError

Base exception raised when a fact could not be collected (e.g. binary absent on the remote host, or the fact was skipped by a condition).

MissingCommandError ¶

MissingCommandError(command: str)

Bases: FactNotCollected

Exception raised when requires_command specifies a binary that is not present on the remote host. The fact returns its default() value instead of raising, unless explicitly configured otherwise.

FactPreconditionError ¶

FactPreconditionError(fact_cls: type, reason: str)

Bases: FactNotCollected

Exception raised when a fact's check_preconditions() returns a reason string (e.g. a kernel module is not loaded). Like MissingCommandError, this is silenced during the prepare phase and re-raised during execute.

DeployError ¶

Bases: PyinfraError

User exception for raising in deploys or sub deploys.

InventoryError ¶

Bases: PyinfraError

Exception raised for inventory related errors.

NoConnectorError ¶

Bases: PyinfraError, ValueError

Raised when a requested connector is missing.

NoHostError ¶

Bases: PyinfraError, KeyError

Raised when an inventory is missing a host.

NoGroupError ¶

Bases: PyinfraError, KeyError

Raised when an inventory is missing a group.

ConnectorDataTypeError ¶

Bases: PyinfraError, TypeError

Raised when host connector data has invalid types.

ArgumentTypeError ¶

Bases: PyinfraError, TypeError

Raised when global arguments are passed with invalid types.

facts ¶

The pyinfra facts API. Facts enable pyinfra to collect remote server state which is used to "diff" with the desired state, producing the final commands required for a deploy.

Note that the facts API does not use the global currently in context host so it's possible to call facts on hosts out of context (ie give me the IP of this other host B while I operate on this host A).

FactBase ¶

Bases: Generic[T]

requires_command ¶

requires_command(*args, **kwargs) -> str | None

Return the binary name that must exist on the remote host for this fact to run. If the binary is absent the fact returns its default() value silently.

check_preconditions ¶

check_preconditions(state: State, host: Host) -> str | None

Check that this fact's prerequisites are satisfied before running.

Override this method to call host.get_fact(...) and return:

None (or no return) — all prerequisites satisfied, proceed normally
"reason message" — prerequisite not satisfied with explanation

The framework handles raising FactPreconditionError and phase-awareness automatically; fact authors never need to import exception classes.

default `staticmethod` ¶

default() -> T

Set the default attribute to be a type (eg list/dict).

hiddenvalue ¶

HiddenValue ¶

HiddenValue(content='', masked_value='*MASKED*')

A class to contain a hidden value To retrieve the real value use .unmask()

host ¶

HostData ¶

HostData(host: Host, *datas)

Combines multiple AttrData's to search for attributes.

Host ¶

Host(
    name: str,
    inventory: Inventory,
    groups,
    connector_cls=None,
)

Represents a target host. Thin class that links up to facts and host/group data.

noop ¶

noop(description: str) -> None

Log a description for a noop operation.

deploy ¶

deploy(
    name: str,
    kwargs: AllArguments | None,
    data: dict | None,
    in_deploy: bool = True,
)

Wraps a group of operations as a deploy, this should not be used directly, instead use pyinfra.api.deploy.deploy.

get_temp_filename ¶

get_temp_filename(
    hash_key: str | None = None,
    hash_filename: bool = True,
    *,
    temp_directory: str | None = None,
)

Generate a temporary filename for this deploy.

get_fact ¶

get_fact(
    name_or_cls: type[FactBase[T]], *args, **kwargs
) -> T

get_fact(
    name_or_cls: type[ShortFactBase[T]], *args, **kwargs
) -> T

get_fact(name_or_cls, *args, **kwargs)

Get a fact for this host, reading from the cache if present.

connect ¶

connect(
    reason=None,
    show_errors: bool = True,
    raise_exceptions: bool = False,
)

Connect to the host using it's configured connector.

disconnect ¶

disconnect() -> None

Disconnect from the host using it's configured connector.

run_shell_command ¶

run_shell_command(
    *args, **kwargs
) -> tuple[bool, CommandOutput]

Low level method to execute a shell command on the host via it's configured connector.

put_file ¶

put_file(*args, **kwargs) -> bool

Low level method to upload a file to the host via it's configured connector.

get_file ¶

get_file(*args, **kwargs) -> bool

Low level method to download a file from the host via it's configured connector.

inventory ¶

Inventory ¶

Inventory(names_data, override_data=None, **groups)

Represents a collection of target hosts. Stores and provides access to group data, host data and default data for these hosts.

Parameters:

Name	Description	Default
`names_data`	tuple of `(names, data)`	required
`override_data`	dictionary of data overrides	`None`
`ssh_*`	deprecated, use `override_data.ssh_*`	required
`winrm_*`	deprecated, use `override_data.winrm_*`	required
`**groups`	map of group name -> `(names, data)`	`{}`

len ¶

__len__() -> int

Returns the number of inventory hosts.

iter ¶

__iter__() -> Iterator[Host]

Iterates over all inventory hosts.

get_active_hosts ¶

get_active_hosts() -> list[Host]

Iterates over active inventory hosts.

len_active_hosts ¶

len_active_hosts() -> int

Returns the number of active inventory hosts.

iter_activated_hosts ¶

iter_activated_hosts() -> Iterator[Host]

Iterates over activated inventory hosts.

len_activated_hosts ¶

len_activated_hosts() -> int

Returns the number of activated inventory hosts.

get_host ¶

get_host(name: str, default=NoHostError) -> Host

Get a single host by name.

get_group ¶

get_group(name: str, default=NoGroupError) -> list[Host]

Get a list of hosts belonging to a group.

get_data ¶

get_data()

Get the base/all data attached to this inventory.

get_override_data ¶

get_override_data()

Get override data for this inventory.

get_host_data ¶

get_host_data(hostname: str)

Get data for a single host in this inventory.

get_group_data ¶

get_group_data(group)

Get data for a single group in this inventory.

get_groups_data ¶

get_groups_data(groups)

Gets aggregated data from a list of groups. Vars are collected in order so, for any groups which define the same var twice, the last group's value will hold.

metadata ¶

Support parsing pyinfra-metadata.toml

Currently just parses plugins and their metadata.

Tag ¶

Bases: BaseModel

Representation of a plugin tag.

Plugin ¶

Bases: BaseModel

Representation of a pyinfra plugin.

parse_plugins ¶

parse_plugins(metadata_text: str) -> list[Plugin]

Given the contents of a pyinfra-metadata.toml parse out the plugins.

operation ¶

Operations are the core of pyinfra. The @operation wrapper intercepts calls to the function and instead diff against the remote server, outputting commands to the deploy state. This is then run later by pyinfra's __main__ or the :doc:./pyinfra.api.operations module.

OperationMeta ¶

OperationMeta(hash, is_change: bool | None)

was_retried `property` ¶

was_retried: bool

Returns whether this operation was retried at least once.

retry_succeeded `property` ¶

retry_succeeded: bool | None

Returns whether this operation succeeded after retries. Returns None if the operation was not retried.

repr ¶

__repr__() -> str

Return Operation object as a string.

get_retry_info ¶

get_retry_info() -> dict[str, Any]

Returns a dictionary with all retry-related information.

add_op ¶

add_op(state: State, op_func, *args, **kwargs)

Prepare & add an operation to pyinfra.state by executing it on all hosts.

Parameters:

Name	Type	Description	Default
`state`	``pyinfra.api.State`` obj	the deploy state to add the operation	required
`to op_func`	`function`	the operation function from one of the modules,	required
`args/kwargs`		passed to the operation function	required

operation ¶

operation(
    is_idempotent: bool = True,
    idempotent_notice: str | None = None,
    is_deprecated: bool = False,
    deprecated_for: str | None = None,
    _set_in_op: bool = True,
) -> Callable[
    [Callable[P, Generator]], PyinfraOperation[P]
]

Decorator that takes a simple module function and turn it into the internal operation representation that consists of a list of commands + options (sudo, (sudo|su)_user, env).

operations ¶

run_ops ¶

run_ops(
    state: State,
    serial: bool = False,
    no_wait: bool = False,
)

Runs all operations across all servers in a configurable manner.

Parameters:

Name	Type	Description	Default
`state`	``pyinfra.api.State`` obj	the deploy state to execute	required
`serial`	`boolean`	whether to run operations host by host	`False`
`no_wait`	`boolean`	whether to wait for all hosts between operations	`False`

output ¶

Pluggable output formatting for pyinfra.

Provides format_text and echo functions that default to plain-text no-ops, allowing the API layer to work without any CLI dependency. The CLI layer replaces them at startup via set_formatter and set_echo.

format_text ¶

format_text(text: str, *args: Any, **kwargs: Any) -> str

Format text with optional styling (color, bold, etc.).

Mirrors click.style signature. Accepts positional fg argument for compatibility with click.style("text", "red").

echo ¶

echo(message: Any = None, **kwargs: Any) -> None

Echo a message. Mirrors click.echo signature.

set_formatter ¶

set_formatter(func: Callable[..., str]) -> None

Replace the default formatter (e.g. with click.style).

set_echo ¶

set_echo(func: Callable[..., None]) -> None

Replace the default echo function (e.g. with click.echo).

is_output_active ¶

is_output_active() -> bool

Return True if a non-default echo function has been installed.

state ¶

State ¶

State(
    inventory: Inventory | None = None,
    config: Config | None = None,
    check_for_changes: bool = True,
    **kwargs,
)

Manages state for a pyinfra deploy.

Initializes the state, the main Pyinfra

Parameters:

Name	Type	Description	Default
`inventory`	`Optional[Inventory]`	The inventory. Defaults to None.	`None`
`config`	`Optional[Config]`	The config object. Defaults to None.	`None`

activate_host ¶

activate_host(host: Host)

Flag a host as active.

fail_hosts ¶

fail_hosts(hosts_to_fail, activated_count=None)

Flag a set of hosts as failed, error for config.FAIL_PERCENT.

is_host_in_limit ¶

is_host_in_limit(host: Host)

Returns a boolean indicating if the host is within the current state limit.

util ¶

get_file_io ¶

get_file_io(filename_or_io: str | IO, mode: str = 'rb')

Given either a filename or an existing IO object, this context processor will open and close filenames, and leave IO objects alone.

get_template ¶

get_template(
    filename_or_io: str | IO,
    jinja_env_kwargs: dict[str, Any] | None = None,
) -> Template

Gets a jinja2 Template object for the input filename or string, with caching based on the filename of the template, or the SHA1 of the input string.

sha1_hash ¶

sha1_hash(string: str) -> str

Return the SHA1 of the input string.

make_hash ¶

make_hash(obj)

Make a hash from an arbitrary nested dictionary, list, tuple or set, used to generate ID's for operations based on their name & arguments.

get_path_permissions_mode ¶

get_path_permissions_mode(pathname: str)

Get the permissions (bits) of a path as an integer.

API Reference¶

Core API¶

pyinfra.api ¶

arguments ¶

all_global_arguments ¶

pop_global_arguments ¶

command ¶

make_formatted_string_command ¶

config ¶

Config ¶

connect ¶

connect_all ¶

disconnect_all ¶

deploy ¶

add_deploy ¶

deploy ¶

exceptions ¶

PyinfraError ¶

ConnectError ¶

FactError ¶

FactTypeError ¶

FactValueError ¶

FactProcessError ¶

OperationError ¶

OperationTypeError ¶

OperationValueError ¶

NestedOperationError ¶

FactNotCollected ¶

MissingCommandError ¶

FactPreconditionError ¶

DeployError ¶

InventoryError ¶

NoConnectorError ¶

NoHostError ¶

NoGroupError ¶

ConnectorDataTypeError ¶

ArgumentTypeError ¶

facts ¶

FactBase ¶

requires_command ¶

check_preconditions ¶

default staticmethod ¶

hiddenvalue ¶

HiddenValue ¶

host ¶

HostData ¶

Host ¶

noop ¶

deploy ¶

get_temp_filename ¶

get_fact ¶

connect ¶

disconnect ¶

run_shell_command ¶

put_file ¶

get_file ¶

inventory ¶

Inventory ¶

__len__ ¶

__iter__ ¶

get_active_hosts ¶

len_active_hosts ¶

iter_activated_hosts ¶

len_activated_hosts ¶

get_host ¶

get_group ¶

get_data ¶

get_override_data ¶

get_host_data ¶

get_group_data ¶

get_groups_data ¶

metadata ¶

Tag ¶

Plugin ¶

parse_plugins ¶

operation ¶

OperationMeta ¶

was_retried property ¶

retry_succeeded property ¶

__repr__ ¶

default `staticmethod` ¶

len ¶

iter ¶

was_retried `property` ¶

retry_succeeded `property` ¶

repr ¶