runway.cfngin.plan module

CFNgin plan, plan components, and functions for interacting with a plan.

class runway.cfngin.plan.Graph

Bases: object

Graph represents a graph of steps.

The Graph helps organize the steps needed to execute a particular action for a set of runway.cfngin.stack.Stack objects. When initialized with a set of steps, it will first build a Directed Acyclic Graph from the steps and their dependencies.

Example

>>> dag = DAG()
>>> a = Step("a", fn=deploy)
>>> b = Step("b", fn=deploy)
>>> dag.add_step(a)
>>> dag.add_step(b)
>>> dag.connect(a, b)

__init__(steps: dict[str, Step] | None = None, dag: DAG | None = None) → None

Instantiate class.

Parameters:

• steps – Dict with key of step name and value of Step for steps to initialize the Graph with. Note that if this is provided, a pre-configured runway.cfngin.dag.DAG that already includes these steps should also be provided..

• dag – An optional runway.cfngin.dag.DAG object. If one is not provided, a new one will be initialized.

__str__() → str

Object displayed as a string.

add_step(step: Step, add_dependencies: bool = False, add_dependents: bool = False) → None

Add a step to the graph.

Parameters:

• step – The step to be added.

• add_dependencies – Connect steps that need to be completed before this step.

• add_dependents – Connect steps that require this step.

add_step_if_not_exists(step: Step, add_dependencies: bool = False, add_dependents: bool = False) → None

Try to add a step to the graph.

Can be used when failure to add is acceptable.

Parameters:

• step – The step to be added.

• add_dependencies – Connect steps that need to be completed before this step.

• add_dependents – Connect steps that require this step.

add_steps(steps: list[Step]) → None

Add a list of steps.

Parameters:

steps – The step to be added.

connect(step: str, dep: str) → None

Connect a dependency to a step.

Parameters:

• step – Step name to add a dependency to.

• dep – Name of dependent step.

downstream(step_name: str) → list[Step]

Return the direct dependencies of the given step.

dumps(indent: int | None = None) → str

Output the graph as a json serialized string for storage.

Parameters:

indent – Number of spaces for each indentation.

filtered(step_names: list[str]) → Graph

Return a “filtered” version of this graph.

Parameters:

step_names – Steps to filter.

classmethod from_dict(graph_dict: dict[str, list[str]] | dict[str, set[str]] | OrderedDict[str, set[str]], context: CfnginContext) → Graph

Create a Graph from a graph dict.

Parameters:

• graph_dict – The dictionary used to create the graph.

• context – Required to init stacks.

classmethod from_steps(steps: list[Step]) → Graph

Create a Graph from Steps.

Parameters:

steps – Steps used to create the graph.

pop(step: Step, default: Any = None) → Any

Remove a step from the graph.

Parameters:

• step – The step to remove from the graph.

• default – Returned if the step could not be popped

to_dict() → OrderedDict[str, set[str]]

Return the underlying DAG as a dictionary.

topological_sort() → list[Step]

Perform a topological sort of the underlying DAG.

transitive_reduction() → None

Perform a transitive reduction on the underlying DAG.

The transitive reduction of a graph is a graph with as few edges as possible with the same reachability as the original graph.

See https://en.wikipedia.org/wiki/Transitive_reduction

transposed() → Graph

Return a “transposed” version of this graph.

Useful for walking in reverse.

walk(walker: Callable[[DAG, Callable[[str], Any]], Any], walk_func: Callable[[Step], Any]) → Any

Walk the steps of the graph.

Parameters:

• walker – Function used to walk the steps.

• walk_func – Function called with a Step as the only argument for each step of the plan.

class runway.cfngin.plan.Plan

Bases: object

A convenience class for working on a Graph.

context

Context object.

Type:

CfnginContext | None

description

Plan description.

Type:

str

graph

Graph of the plan.

Type:

Graph

id

UUID for the plan.

Type:

uuid.UUID

reverse

The graph has been transposed for walking in reverse.

Type:

bool

require_unlocked

Require the persistent graph to be unlocked before executing steps.

Type:

bool

__init__(description: str, graph: Graph, context: CfnginContext | None = None, reverse: bool = False, require_unlocked: bool = True) → None

Initialize class.

Parameters:

• description – Description of what the plan is going to do.

• graph – Local graph used for the plan.

• context – Context object.

• reverse – Transpose the graph for walking in reverse.

• require_unlocked – Require the persistent graph to be unlocked before executing steps.

dump(*, directory: str, context: CfnginContext, provider: Provider | None = None) → Any

Output the rendered blueprint for all stacks in the plan.

Parameters:

• directory – Directory where files will be created.

• context – Current CFNgin context.

• provider – Provider to use when resolving the blueprints.

execute(*args: Any, **kwargs: Any) → None

Walk each step in the underlying graph.

Raises:

• PersistentGraphLocked – Raised if the persistent graph is locked prior to execution and this session did not lock it.

• PlanFailed – Raised if any of the steps fail.

keys() → list[str]

Return a list of all step names.

property lock_code: str

Code to lock/unlock the persistent graph.

outline(level: int = 20, message: str = '') → None

Print an outline of the actions the plan is going to take.

The outline will represent the rough ordering of the steps that will be taken.

Parameters:

• level – a valid log level that should be used to log the outline

• message – a message that will be logged to the user after the outline has been logged.

property step_names: list[str]

Return a list of all step names.

property steps: list[Step]

Return a list of all steps in the plan.

walk(walker: Callable[[...], Any]) → Any

Walk each step in the underlying graph, in topological order.

Parameters:

walker – a walker function to be passed to runway.cfngin.dag.DAG to walk the graph.

class runway.cfngin.plan.Step

Bases: object

State machine for executing generic actions related to stacks.

fn

Function to run to execute the step. This function will be ran multiple times until the step is “done”.

Type:

Callable[…, Any] | None

last_updated

Time when the step was last updated.

Type:

float

logger

Logger for logging messages about the step.

Type:

PrefixAdaptor

stack

the stack associated with this step

Type:

Stack

status

The status of step.

Type:

Status

watch_func

Function that will be called to “tail” the step action.

Type:

Callable[…, Any] | None

__init__(stack: Stack, *, fn: Callable[[...], Any] | None = None, watch_func: Callable[[...], Any] | None = None) → None

Instantiate class.

Parameters:

• stack – The stack associated with this step

• fn – Function to run to execute the step. This function will be ran multiple times until the step is “done”.

• watch_func – Function that will be called to “tail” the step action.

__repr__() → str

Object represented as a string.

__str__() → str

Object displayed as a string.

complete() → None

Shortcut for set_status(COMPLETE).

property completed: bool

Return True if the step is in a COMPLETE state.

property done: bool

Return True if the step is finished.

To be True, status must be either COMPLETE, SKIPPED or FAILED)

property failed: bool

Return True if the step is in a FAILED state.

classmethod from_persistent_graph(graph_dict: dict[str, list[str]] | dict[str, set[str]] | OrderedDict[str, set[str]], context: CfnginContext, fn: Callable[..., Status] | None = None, watch_func: Callable[..., Any] | None = None) → list[Step]

Create a steps for a persistent graph dict.

Parameters:

• graph_dict – A graph dict.

• context – Context object. Required to initialize a “fake” runway.cfngin.stack.Stack.

• requires – Stacks that this stack depends on.

• fn – The function to run to execute the step. This function will be ran multiple times until the step is “done”.

• watch_func – an optional function that will be called to “tail” the step action.

classmethod from_stack_name(stack_name: str, context: CfnginContext, requires: list[str] | set[str] | None = None, fn: Callable[..., Status] | None = None, watch_func: Callable[..., Any] | None = None) → Step

Create a step using only a stack name.

Parameters:

• stack_name – Name of a CloudFormation stack.

• context – Context object. Required to initialize a “fake” runway.cfngin.stack.Stack.

• requires – Stacks that this stack depends on.

• fn – The function to run to execute the step. This function will be ran multiple times until the step is “done”.

• watch_func – an optional function that will be called to “tail” the step action.

log_step() → None

Construct a log message for a set and log it to the UI.

property name: str

Name of the step.

This is equal to the name of the stack it operates on.

property ok: bool

Return True if the step is finished (either COMPLETE or SKIPPED).

property required_by: set[str]

Return a list of step names that depend on this step.

property requires: set[str]

Return a list of step names this step depends on.

run() → bool

Run this step until it has completed or been skipped.

set_status(status: Status) → None

Set the current step’s status.

Parameters:

status – The status to set the step to.

skip() → None

Shortcut for set_status(SKIPPED).

property skipped: bool

Return True if the step is in a SKIPPED state.

submit() → None

Shortcut for set_status(SUBMITTED).

property submitted: bool

Return True if the step is SUBMITTED, COMPLETE, or SKIPPED.

runway.cfngin.plan.json_serial(obj: set[_T]) → list[_T]

runway.cfngin.plan.json_serial(obj: dict[Any, Any] | int | list[Any] | str) → NoReturn

Serialize json.

Parameters:

obj – A python object.

Example

json.dumps(data, default=json_serial)

runway.cfngin.plan.merge_graphs(graph1: Graph, graph2: Graph) → Graph

Combine two Graphs into one, retaining steps.

Parameters:

• graph1 – Graph that graph2 will be merged into.

• graph2 – Graph that will be merged into graph1.