runway.cfngin.plan module

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

runway.cfngin.plan.json_serial(obj: Set[_T]) → List[_T]

runway.cfngin.plan.json_serial(obj: Union[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: runway.cfngin.plan.Graph, graph2: runway.cfngin.plan.Graph) → runway.cfngin.plan.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.

class runway.cfngin.plan.Step(stack: Stack, *, fn: Optional[Callable[…, Any]] = None, watch_func: Optional[Callable[…, Any]] = None)

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

Optional[Callable[.., Any]]

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

Optional[Callable[.., Any]]

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.

run() → bool

Run this step until it has completed or been skipped.

property name

Name of the step.

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

property requires

Return a list of step names this step depends on.

property required_by

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

property completed

Return True if the step is in a COMPLETE state.

property skipped

Return True if the step is in a SKIPPED state.

property failed

Return True if the step is in a FAILED state.

property done

Return True if the step is finished.

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

property ok

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

property submitted

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

set_status(status: Status) → None

Set the current step’s status.

Parameters

status – The status to set the step to.

complete() → None

Shortcut for set_status(COMPLETE).

log_step() → None

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

skip() → None

Shortcut for set_status(SKIPPED).

submit() → None

Shortcut for set_status(SUBMITTED).

classmethod from_stack_name(stack_name: str, context: CfnginContext, requires: Optional[Union[List[str], Set[str]]] = None, fn: Optional[Callable[…, Status]] = None, watch_func: Optional[Callable[…, Any]] = 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.

classmethod from_persistent_graph(graph_dict: Union[Dict[str, List[str]], Dict[str, Set[str]], OrderedDict[str, Set[str]]], context: CfnginContext, fn: Optional[Callable[…, Status]] = None, watch_func: Optional[Callable[…, Any]] = 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.

__repr__() → str

Object represented as a string.

__str__() → str

Object displayed as a string.

class runway.cfngin.plan.Graph(steps: Optional[Dict[str, Step]] = None, dag: Optional[DAG] = None)

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)

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.

add_step(step: runway.cfngin.plan.Step, add_dependencies: bool = False, add_dependants: 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_dependants – Connect steps that require this step.

add_step_if_not_exists(step: runway.cfngin.plan.Step, add_dependencies: bool = False, add_dependants: 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_dependants – Connect steps that require this step.

add_steps(steps: List[Step]) → None

Add a list of steps.

Parameters

steps – The step to be added.

pop(step: runway.cfngin.plan.Step, default: Optional[typing.Any] = None) → typing.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

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.

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

walk(walker: Callable[[runway.cfngin.dag.DAG, Callable[[str], typing.Any]], typing.Any], walk_func: Callable[[runway.cfngin.plan.Step], typing.Any]) → typing.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.

downstream(step_name: str) → List[Step]

Return the direct dependencies of the given step.

transposed() → runway.cfngin.plan.Graph

Return a “transposed” version of this graph.

Useful for walking in reverse.

filtered(step_names: List[str]) → Graph

Return a “filtered” version of this graph.

Parameters

step_names – Steps to filter.

topological_sort() → List[Step]

Perform a topological sort of the underlying DAG.

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

Return the underlying DAG as a dictionary.

dumps(indent: Optional[int] = None) → str

Output the graph as a json seralized string for storage.

Parameters

indent – Number of spaces for each indentation.

classmethod from_dict(graph_dict: Union[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.

__str__() → str

Object displayed as a string.

class runway.cfngin.plan.Plan(description: str, graph: Graph, context: Optional[CfnginContext] = None, reverse: bool = False, require_unlocked: bool = True)

Bases: object

A convenience class for working on a Graph.

context

Context object.

Type

Optional[CfnginContext]

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

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.

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

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.

dump(*, directory: str, context: CfnginContext, provider: Optional[Provider] = 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: typing.Any, **kwargs: typing.Any)

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.

walk(walker: Callable[[…], typing.Any]) → typing.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.

property lock_code

Code to lock/unlock the persistent graph.

property steps

Return a list of all steps in the plan.

property step_names

Return a list of all step names.

keys() → List[str]

Return a list of all step names.