runway.cfngin.plan module

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

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

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[source]

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[source]

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]]

__init__(stack: runway.cfngin.stack.Stack, *, fn: Optional[Callable[[...], Any]] = None, watch_func: Optional[Callable[[...], Any]] = None) → None[source]

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[source]: Run this step until it has completed or been skipped.

property name: str

Name of the step.

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

property requires: Set[str]: Return a list of step names this step depends on.

property required_by: Set[str]: Return a list of step names that depend on this step.

property completed: bool: Return True if the step is in a COMPLETE state.

property skipped: bool: Return True if the step is in a SKIPPED state.

property failed: bool: Return True if the step is in a FAILED state.

property done: bool

Return True if the step is finished.

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

property ok: bool: Return True if the step is finished (either COMPLETE or SKIPPED).

property submitted: bool: Return True if the step is SUBMITTED, COMPLETE, or SKIPPED.

set_status(status: Status) → None[source]

Set the current step’s status.

Parameters: status – The status to set the step to.

complete() → None[source]: Shortcut for set_status(COMPLETE).

log_step() → None[source]: Construct a log message for a set and log it to the UI.

skip() → None[source]: Shortcut for set_status(SKIPPED).

submit() → None[source]: 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[source]

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][source]

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[source]: Object represented as a string.

__str__() → str[source]: Object displayed as a string.

__new__(**kwargs)

class runway.cfngin.plan.Graph[source]

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: Optional[Dict[str, runway.cfngin.plan.Step]] = None, dag: Optional[runway.cfngin.dag.DAG] = None) → None[source]

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_dependents: bool = False) → None[source]

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: runway.cfngin.plan.Step, add_dependencies: bool = False, add_dependents: bool = False) → None[source]

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[runway.cfngin.plan.Step]) → None[source]

Add a list of steps.

Parameters: steps – The step to be added.

pop(step: runway.cfngin.plan.Step, default: Any = None) → Any[source]

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[source]

Connect a dependency to a step.

Parameters

step – Step name to add a dependency to.
dep – Name of dependent step.

transitive_reduction() → None[source]

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], Any]], Any], walk_func: Callable[[runway.cfngin.plan.Step], Any]) → Any[source]

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[runway.cfngin.plan.Step][source]: Return the direct dependencies of the given step.

transposed() → runway.cfngin.plan.Graph[source]

Return a “transposed” version of this graph.

Useful for walking in reverse.

filtered(step_names: List[str]) → runway.cfngin.plan.Graph[source]

Return a “filtered” version of this graph.

Parameters: step_names – Steps to filter.

topological_sort() → List[runway.cfngin.plan.Step][source]: Perform a topological sort of the underlying DAG.

to_dict() → OrderedDict[str, Set[str]][source]: Return the underlying DAG as a dictionary.

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

Output the graph as a json serialized 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: runway.context.CfnginContext) → runway.cfngin.plan.Graph[source]

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[runway.cfngin.plan.Step]) → runway.cfngin.plan.Graph[source]

Create a Graph from Steps.

Parameters: steps – Steps used to create the graph.

__str__() → str[source]: Object displayed as a string.

__new__(**kwargs)

class runway.cfngin.plan.Plan[source]

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

__init__(description: str, graph: runway.cfngin.plan.Graph, context: typing.Optional[<sphinx.util.inspect.TypeAliasForwardRef object at 0x7f570523a360>] = None, reverse: bool = False, require_unlocked: bool = True) → None[source]

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.

__new__(**kwargs)

outline(level: int = 20, message: str = '')[source]

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[source]

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)[source]

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[[...], Any]) → Any[source]

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: str: Code to lock/unlock the persistent graph.

property steps: List[runway.cfngin.plan.Step]: Return a list of all steps in the plan.

property step_names: List[str]: Return a list of all step names.

keys() → List[str][source]: Return a list of all step names.