Idem Scripts

Idem runs can be initialized from a Python script. When possible, use the Idem CLI. However, in some FAAS applications, shelling out isn’t an option; for example, AWS Lambdas must use pure Python.

The following examples show how to run Idem from a pure Python script.

In this example, states are run with the minimum configuration:

import pop.hub
import json

# Create the hub
hub = pop.hub.Hub()
# Add idem's dynamic namespace to the hub, which loads all idem-related subs onto the hub
# I.E. states/exec/tool/acct/etc...
# All python projects in the current python environment that have a conf.py with a DYNE dictionary will be loaded
# They will be loaded onto subs based on the mappings in the DYNE dictionary
hub.pop.sub.add(dyne_name="idem")

# Set up variables to use in the run
run_name = "my_run"
sls_name = "my_sls"
test = False
invert_state = False
acct_profile = "default"
cache_dir = "/dev/null"
runtime = "parallel"
params = []
param_sources = []
esm_plugin = "local"
esm_profile = "default"

# Compile states as a dictionary
my_states = {
    f"{sls_name}.sls": {
        "state_name": {
            "test.nop": [
                {"name": "value"},
                {"kwarg1": "value1"},
            ]
        }
    }
}

# Compile acct data that will be used for the run
acct_data = {
    "profiles": {
        "provider_name": {
            "profile_name": {
                "kwarg_1": "value_1",
                "kwarg_2": "value_2",
            },
            "default": {
                "kwarg_1": "value_1",
                "kwarg_2": "value_2",
            },
        }
    }
}

# Create the event loop
hub.pop.loop.create()
# Run the states
hub.pop.Loop.run_until_complete(
    hub.idem.state.apply(
        name=run_name,
        sls_sources=[f"json://{json.dumps(my_states)}"],
        render="json",
        runtime=runtime,
        subs=["states"],
        cache_dir=cache_dir,
        sls=[sls_name],
        test=test,
        invert_state=invert_state,
        acct_profile=acct_profile,
        acct_data=acct_data,
        managed_state={},
        param_sources=param_sources,
        params=params,
    )
)

# Gather the results from RUNS
results = hub.idem.RUNS[run_name]["running"]
errors = hub.idem.RUNS[run_name]["errors"]

# Do things with the resulting data
print(f"State compile errors: {errors}")
print(f"State run results: {results}")

This example runs states in an esm context:

import pop.hub
import json

# Create the hub
hub = pop.hub.Hub()
# Add idem's dynamic namespace to the hub, which loads all idem-related subs onto the hub
# I.E. states/exec/tool/acct/etc...
# All python projects in the current python environment that have a conf.py with a DYNE dictionary will be loaded
# They will be loaded onto subs based on the mappings in the DYNE dictionary
hub.pop.sub.add(dyne_name="idem")

# Set up variables to use in the run
run_name = "my_run"
sls_name = "my_sls"
test = False
invert_state = False
acct_profile = "default"
cache_dir = "/dev/null"
runtime = "parallel"
params = []
param_sources = []
esm_plugin = "local"
esm_profile = "default"

# Compile states as a dictionary
my_states = {
    f"{sls_name}.sls": {
        "state_name": {
            "test.nop": [
                {"name": "value"},
                {"kwarg1": "value1"},
            ]
        }
    }
}

# Compile acct data that will be used for the run
acct_data = {
    "profiles": {
        "provider_name": {
            "profile_name": {
                "kwarg_1": "value_1",
                "kwarg_2": "value_2",
            },
            "default": {
                "kwarg_1": "value_1",
                "kwarg_2": "value_2",
            },
        }
    }
}


async def start():
    # Configure the context for ESM
    context_manager = hub.idem.managed.context(
        run_name=run_name,
        cache_dir=cache_dir,
        esm_plugin=esm_plugin,
        esm_profile=esm_profile,
        acct_data=acct_data,
    )
    # Run the states in the ESM context
    async with context_manager as state:
        await hub.idem.state.apply(
            name=run_name,
            sls_sources=[f"json://{json.dumps(my_states)}"],
            render="json",
            runtime=runtime,
            subs=["states"],
            cache_dir=cache_dir,
            sls=[sls_name],
            test=test,
            invert_state=invert_state,
            acct_profile=acct_profile,
            acct_data=acct_data,
            managed_state=state,
            param_sources=param_sources,
            params=params,
        )


# Create the event loop
hub.pop.loop.create()
# Run idem in the event loop
hub.pop.Loop.run_until_complete(start)

# Gather the results from RUNS
results = hub.idem.RUNS[run_name]["running"]
errors = hub.idem.RUNS[run_name]["errors"]

# Do things with the resulting data
print(f"State compile errors: {errors}")
print(f"State run results: {results}")

This example runs an exec module with the minimum configuration required:

import pop.hub
import json

# Create the hub
hub = pop.hub.Hub()
# Add idem's dynamic namespace to the hub, which loads all idem-related subs onto the hub
# I.E. states/exec/tool/acct/etc...
# All python projects in the current python environment that have a conf.py with a DYNE dictionary will be loaded
# They will be loaded onto subs based on the mappings in the DYNE dictionary
hub.pop.sub.add(dyne_name="idem")

# Use the run_name as a routing key for exec module events
hub.idem.RUN_NAME = run_name = "my_run"

# Compile acct data that will be used for the run
acct_profile = "default"
acct_data = {
    "profiles": {
        "provider_name": {
            "profile_name": {
                "kwarg_1": "value_1",
                "kwarg_2": "value_2",
            },
            "default": {
                "kwarg_1": "value_1",
                "kwarg_2": "value_2",
            },
        }
    }
}

# positional arguments for the exec module go here.
args = []
# Keyword arguments for teh exec module go here
kwargs = {}

# NOTE: hub and ctx should not be passed to the exec module.
# The hub is implicitly passed and ctx is constructed from acct_data

hub.pop.loop.create()
# Run the exec module in the loop
result = hub.pop.Loop.run_until_complete(
    hub.idem.ex.run(
        "test.ping",
        args=args,
        kwargs=kwargs,
        acct_data=acct_data,
        acct_profile=acct_profile,
    )
)

# Do things with the result
print(result.result)
print(result.comment)
print(result.ret)