Information on how to invoke a deploy script programmatically using the API

[phlummox]

Is your feature request related to a problem? Please describe

It often would be useful to be able to execute a deploy script (or even just a particular function from such a script) programmatically, from Python code. If you're running Python code, then having to invoke PyInfra by using e.g. the subprocess module seems an inefficient way of executing deploy scripts on remote hosts.

(See also this related issue: #989.)

The PyInfra documentation does give some information on how one can invoke individual operations programmatically (https://docs.pyinfra.com/en/2.x/api/index.html), but not a whole deploy script.

Describe the solution you'd like

Would it be possible to provide an example, in the PyInfra documentation or the examples subdirectory, showing how a deploy script can be run programmatically?

It obviously must be possible, since that's exactly what the CLI does using load_deploy_file():

pyinfra/pyinfra_cli/util.py

Lines 191 to 193 in 9256e6a

	def load_deploy_file(state: "State", filename):
	state.current_deploy_filename = filename
	_parallel_load_hosts(state, lambda: exec_file(filename), filename)

And ultimately, load_deploy_file is just using Python's built-in exec() function to execute the deploy script (here).

However, load_deploy_file() seems to rely on some sort of context or state which isn't being set up in the API example at examples/api_deploy.py.

We can try to execute code like the following (based on api_deploy.py):

from gevent import monkey
monkey.patch_all()

import pyinfra.facts
from pyinfra import (config,
                    )
from pyinfra.operations import (
  server,
)
from pyinfra.api import BaseStateCallback, Config, Inventory, State
from pyinfra.api.connect import connect_all

import logging

from pyinfra.api import BaseStateCallback, Config, Inventory, State
from pyinfra.api.connect import connect_all
from pyinfra.api.facts import get_facts
from pyinfra.api.operation import add_op
from pyinfra.api.operations import run_ops
from pyinfra.facts.server import Os
from pyinfra_cli.prints import jsonify

from pyinfra_cli.util import load_deploy_file

class StateCallback(BaseStateCallback):
    def host_connect(self, state, host):
        print("Host connected: {0}".format(host))

    def operation_start(self, state, op_hash):
        print("Start operation: {0}".format(op_hash))

    def operation_end(self, state, op_hash):
        print("End operation: {0}".format(op_hash))

logging.basicConfig(level=logging.INFO)

hosts = ["somehost.example.com"]
inventory = Inventory( (hosts, {}) )

config = Config()
state = State(inventory, config)
state.add_callback_handler(StateCallback())

print("Connecting...")
connect_all(state)

# Adding operations works
print("Generating operations...")
add_op(
    state,
    server.shell,
    _sudo=True,
    commands=["touch myfile"
             ]
)

# But load_deploy_file does not.
# `mydeploy.py` can be completely empty for the purposes of this example -
# it fails to execute due to errors.
load_deploy_file(state, "mydeploy.py")

run_ops(state)
facts = get_facts(state, Os)
print(jsonify(facts, indent=4))

If we try to execute the above script, load_deploy_file() fails with the following exception traceback:

Traceback (most recent call last):
  File "samplescript.py", line 60, in <module>
    load_deploy_file(state, "mydeploy.py")
  File "myenv/lib/python3.10/site-packages/pyinfra_cli/util.py", line 193, in load_deploy_file
    _parallel_load_hosts(state, lambda: exec_file(filename), filename)
  File "myenv/lib/python3.10/site-packages/pyinfra_cli/util.py", line 187, in _parallel_load_hosts
    raise result
  File "myenv/lib/python3.10/site-packages/pyinfra_cli/util.py", line 167, in load_file
    callback()
  File "myenv/lib/python3.10/site-packages/pyinfra_cli/util.py", line 193, in <lambda>
    _parallel_load_hosts(state, lambda: exec_file(filename), filename)
  File "myenv/lib/python3.10/site-packages/pyinfra_cli/util.py", line 44, in exec_file
    state.current_exec_filename = filename
  File "myenv/lib/python3.10/site-packages/pyinfra/context.py", line 56, in __setattr__
    raise TypeError("Cannot assign to context base module")
TypeError: Cannot assign to context base module

Would it be possible in the examples to show what sort of context or state needs to be set up so that a function like load_deploy_file can be run?

Other possibly related issues

#723

[phlummox]

In case this is of use to anyone else encountering the same issues, below I've included code adapted from the pyinfra_cli package (mostly main.py), which can be used to execute deploy scripts (or python code contained in a string) on multiple hosts using the PyInfra API.

There is no parallelization - it would be up to the user to implement that however they wished (PyInfra's CLI uses gevent), to construct an Inventory with additional data or groups, and so on.

I couldn't honestly say I understand the details of what's being done with the State and Config objects that are created, but the code below works for me, with PyInfra version 2.8. Perhaps that's something that could be clarified if an example like this were added to the documentation.

import os
import sys
import logging

from typing import Callable, List, Tuple

from pyinfra import logger, state
from pyinfra.api import Config, Inventory, State
from pyinfra.api.connect import connect_all
from pyinfra.api.operations import run_ops
from pyinfra.context import ctx_config, ctx_host, ctx_state

# Don't write out deploy.pyc etc
sys.dont_write_bytecode = True

# Force line buffering
sys.stdout = os.fdopen(sys.stdout.fileno(), "w", 1)
sys.stderr = os.fdopen(sys.stderr.fileno(), "w", 1)

class LogHandler(logging.Handler):
  "handle log records"

  def emit(self, record):
    try:
      message = self.format(record)
      print(message, file=sys.stderr)
    except Exception:
      self.handleError(record)

def exec_file(filename: str):
  """
  Execute a Python file and optionally return its attributes as a dict.
  """
  state.current_exec_filename = filename

  with open(filename, "r", encoding="utf-8") as f:
    code = f.read()
    compiled_code = compile(code, filename, "exec")

  # Execute the code with locals/globals going into the dict
  globals_dict = {}
  exec(compiled_code, globals_dict)
  return globals_dict

def exec_str(code: str, filename: str):
  """
  Execute a Python module string and optionally return its attributes as a dict.
  """
  filename = "(none)"
  state.current_exec_filename = filename
  compiled_code = compile(code, filename, "exec")

  # Execute the code with locals/globals going into the dict
  globals_dict = {}
  exec(compiled_code, globals_dict)
  return globals_dict

def pyinfra_run(hosts: List[str], operations: List[Tuple[str,Callable]]):
  logger.setLevel(logging.INFO)
  handler = LogHandler()
  logger.addHandler(handler)

  # Setup state, config & inventory
  cwd = os.getcwd()
  state = State()
  state.cwd = cwd
  ctx_state.set(state)

  config = Config()
  config.lock_current_state()

  print("--> Loading inventory...", file=sys.stderr)
  inventory = Inventory( (hosts, {}) )

  # Initialise the state
  state.init(inventory, config)

  # Connect to the hosts & start handling the user commands
  print("--> Connecting to hosts...", file=sys.stderr)
  connect_all(state)

  for i, (filename, callback) in enumerate(operations):
    logger.info(f"Loading: {filename}")

    state.current_op_file_number = i
    state.current_deploy_filename = filename
    for host in state.inventory.iter_active_hosts():
      with ctx_config.use(state.config.copy()):
        with ctx_host.use(host):
          callback()
          logger.info(
            "{0}{1} {2}".format(host.print_prefix, "Ready:", filename),
          )

    # Remove any config changes introduced by the deploy file & any includes
    config.reset_locked_state()

  # if desired: the logic from pyinfra_cli.prints.print_meta could be copied,
  # for pretty-printing of proposed changes
  #print("--> Proposed changes:", file=sys.stderr)
  #print_meta(state)

  print("--> Beginning operation run...", file=sys.stderr)

  run_ops(state, serial=True, no_wait=False)

  # if desired: the logic from pyinfra_cli.prints.print_results could be copied,
  # for pretty-printing of final results
  #print("--> Results:", file=sys.stderr)
  #print_results(state)

if __name__ == "__main__":
  hosts = ["host1.example.com", "host2.example.com"]

  operations = [
      ("mydeploy.py", lambda: exec_file("mydeploy.py")),
      ("(nofile)",    lambda: exec_str("print('hi there')", "(nofile)")),
    ]

  pyinfra_run(hosts, operations)

[phlummox]

Just to add a bit of further information ...

In my version of the above code, I've amended pyinfra_run() to take an Inventory object rather than a list of hostnames - that seems more flexible than just assuming an SSH connector/host.

The code currently in pyinfra_cli.inventory.make_inventory_from_files – here:

pyinfra/pyinfra_cli/inventory.py

Line 153 in 52c3fa6

def make_inventory_from_files(

seems generally useful for quickly turning CLI-style inventory-strings (e.g. @docker/some-ctr-id,@ssh/some.hostname.com) into an Inventory, so I've basically just copy-and-pasted it. (I haven't had a chance to go through the logic and see exactly what it's doing, and if it can be tidied up.)

If you'd be interested in a pull request to add a new example in examples, showing how scripts can be created dynamically at runtime using the API, I'd be happy to create one – I'd be keen to hear what you think might be best practices in such an example before I do so.

[Fizzadar]

Small update on this one: this is on the roadmap for v3. Will initially just be restoring the API docs but I’ll also work on some example code.

[phlummox]

Awesome! I'm looking forward to it :)