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.