Information on how to invoke a deploy script programmatically using the API about pyinfra HOT 4 OPEN

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)