pyinfra.api.state module

class pyinfra.api.state.BaseStateCallback

Bases: object

static host_before_connect(state: State, host: Host)

static host_connect(state: State, host: Host)

static host_connect_error(state: State, host: Host, error)

static host_disconnect(state: State, host: Host)

static operation_end(state: State, op_hash)

static operation_host_error(state: State, host: Host, op_hash)

static operation_host_start(state: State, host: Host, op_hash)

static operation_host_success(state: State, host: Host, op_hash)

static operation_start(state: State, op_hash)

class pyinfra.api.state.State(inventory: 'Inventory' | None = None, config: 'Config' | None = None, check_for_changes: bool = True, **kwargs)

Bases: object

Manages state for a pyinfra deploy.

activate_host(host: Host)

Flag a host as active.

add_callback_handler(handler)

check_for_changes: bool = True

config: Config

current_deploy_filename: str | None = None

current_exec_filename: str | None = None

current_op_file_number: int = 0

current_stage: StateStage = 1

cwd: str | None = None

fail_hosts(hosts_to_fail, activated_count=None)

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

get_meta_for_host(host: Host) → StateHostMeta

get_op_data_for_host(host: Host, op_hash: str) → StateOperationHostData

get_op_meta(op_hash: str) → StateOperationMeta

get_op_order()

get_results_for_host(host: Host) → StateHostResults

get_warning_counter() → int

increment_warning_counter() → None

init(inventory: Inventory, config: 'Config' | None, initial_limit=None)

initialised: bool = False

inventory: Inventory

is_executing: bool = False

is_host_in_limit(host: Host)

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

pool: Pool

print_fact_info: bool = False

print_fact_input: bool = False

print_fact_output: bool = False

print_input: bool = False

print_noop_info: bool = False

print_output: bool = False

set_op_data_for_host(host: Host, op_hash: str, op_data: StateOperationHostData)

set_stage(stage: StateStage) → None

should_check_for_changes()

should_raise_failed_hosts: Callable[['State'], bool] | None = None

stage_warnings: dict[StateStage, int] = {}

trigger_callbacks(method_name: str, *args, **kwargs)

class pyinfra.api.state.StateHostMeta

Bases: object

op_hashes: set[str]

ops = 0

ops_change = 0

ops_no_change = 0

class pyinfra.api.state.StateHostResults

Bases: object

error_ops = 0

ignored_error_ops = 0

ops = 0

partial_ops = 0

success_ops = 0

class pyinfra.api.state.StateOperationHostData(command_generator: "Callable[[], Iterator['PyinfraCommand']]", global_arguments: "'AllArguments'", operation_meta: "'OperationMeta'", parent_op_hash: 'Optional[str]' = None)

Bases: object

command_generator: Callable[[], Iterator['PyinfraCommand']]

global_arguments: AllArguments

operation_meta: OperationMeta

parent_op_hash: str | None = None

class pyinfra.api.state.StateOperationMeta(op_order: tuple[int, ...])

Bases: object

args: list[str]

global_arguments: AllArguments

names: set[str]

op_order: tuple[int, ...]

class pyinfra.api.state.StateStage(*values)

Bases: IntEnum

Connect = 2

Disconnect = 5

Execute = 4

Prepare = 3

Setup = 1