Python API

API for writing configurators

class unfurl.configurator.Configurator(configurationSpec: unfurl.planrequests.ConfigurationSpec)
Parameters

configurationSpec (unfurl.planrequests.ConfigurationSpec) –

Return type

None

attribute_output_metadata_key: Optional[str] = None
can_dry_run(task: unfurl.configurator.TaskView) → bool

Returns whether this configurator can handle a dry-runs for the given task. (And should check TaskView.dry_run in during run().

Parameters

task (TaskView) –

Returns

bool

Return type

bool

can_run(task: unfurl.configurator.TaskView) → Union[bool, str]

Return whether or not the configurator can execute the given task depending on if this configurator support the requested action and parameters and given the current state of the target instance?

Parameters

task (TaskView) –

Returns

Should return True or a message describing why the task couldn’t be run.

Return type

(bool or str)

check_digest(task: TaskView, changeset: ChangeRecordRecord) → bool

Examine the previous ChangeRecord generated by the previous time this operation was performed on the target instance and return whether it should be rerun or not.

The default implementation recalculates the digest of input parameters that were accessed in the previous run.

Parameters

  • task (TaskView) –

  • changest (ChangeRecord) –

  • changeset (ChangeRecordRecord) –

Returns

True if configuration’s digest has changed, False if it is the same.

Return type

bool

exclude_from_digest: Tuple[str, ...] = ()
get_generator(task: unfurl.configurator.TaskView) → Generator
Parameters

task (unfurl.configurator.TaskView) –

Return type

Generator

render(task: unfurl.configurator.TaskView) → Any

This method is called is called during the planning phase to give the configurator an opportunity to do early validation and error detection and generate any plan information or configuration files that the user may want to review before the running the deployment task.

Property access and writes will be tracked and used to establish dynamic dependencies between instances so the plan can be ordered properly. Any updates made to instances maybe reverted if it has dependencies on attributes that might be changed later in the plan, so this method should be idempotent.

Returns

The value returned here will subsequently be available as task.rendered

Parameters

task (unfurl.configurator.TaskView) –

Return type

Any

run(task: unfurl.configurator.TaskView) → Generator

This should perform the operation specified in the ConfigurationSpec on the task.target.

Parameters

task (TaskView) –

Yields

Should yield either a JobRequest, TaskRequest or a ConfiguratorResult when done

Return type

Generator

save_digest(task: unfurl.configurator.TaskView) → dict

Generate a compact, deterministic representation of the current configuration. This is saved in the job log and used by check_digest in subsequent jobs to determine if the configuration changed the operation needs to be re-run.

The default implementation calculates a SHA1 digest of the values of the inputs that where accessed while that task was run, with the exception of the input parameters listed in exclude_from_digest.

Parameters

task (TaskView) –

Returns

A dictionary whose keys are strings that start with “digest”

Return type

dict

classmethod set_config_spec_args(kw: dict, target: unfurl.runtime.EntityInstance) → dict
Parameters

  • kw (dict) –

  • target (unfurl.runtime.EntityInstance) –

Return type

dict

short_name: Optional[str] = None

shortName can be used to customize the “short name” of the configurator as an alternative to using the full name (“module.class”) when setting the implementation on an operation. (Titlecase recommended)

should_run(task: unfurl.configurator.TaskView) → Union[bool, unfurl.support.Priority]

Does this configuration need to be run?

Parameters

task (unfurl.configurator.TaskView) –

Return type

Union[bool, unfurl.support.Priority]

class unfurl.configurator.JobRequest(resources: List[unfurl.runtime.EntityInstance], errors: Optional[Sequence[unfurl.util.UnfurlError]] = None, update=None)

Yield this to run a child job.

Parameters
get_instance_specs()
property name
property root
set_error(msg: str)
Parameters

msg (str) –

property target
class unfurl.configurator.TaskRequest(configSpec: unfurl.planrequests.ConfigurationSpec, target: unfurl.runtime.EntityInstance, reason: str, persist: bool = False, required: Optional[bool] = None, startState: Optional[unfurl.support.NodeState] = None)

Yield this to run a child task. (see unfurl.configurator.TaskView.create_sub_task())

Parameters

  • configSpec (unfurl.planrequests.ConfigurationSpec) –

  • target (unfurl.runtime.EntityInstance) –

  • reason (str) –

  • persist (bool) –

  • required (Optional[bool]) –

  • startState (Optional[unfurl.support.NodeState]) –

property completed: bool

bool(x) -> bool

Returns True when the argument x is true, False otherwise. The builtins True and False are the only two instances of the class bool. The class bool is a subclass of the class int, and cannot be subclassed.

finish_workflow() → None
Return type

None

get_operation_artifacts()
property name
reassign_final_for_workflow() → Optional[unfurl.planrequests.TaskRequest]
Return type

Optional[unfurl.planrequests.TaskRequest]

class unfurl.configurator.TaskView(manifest: Manifest, configSpec: unfurl.planrequests.ConfigurationSpec, target: unfurl.runtime.EntityInstance, reason: Optional[str] = None, dependencies: Optional[List[Operational]] = None)

The interface presented to configurators.

The following public attributes are available:

Parameters

  • manifest (Manifest) –

  • configSpec (unfurl.planrequests.ConfigurationSpec) –

  • target (unfurl.runtime.EntityInstance) –

  • reason (Optional[str]) –

  • dependencies (Optional[List[Operational]]) –

Return type

None

target

The instance this task is operating on.

reason

The reason this operation was planned. See Reason

Type

str

cwd

Current working directory

Type

str

dry_run

Dry run only

Type

bool

verbose

Verbosity level set for this job (-1 error, 0 normal, 1 verbose, 2 debug)

Type

int

add_dependency(expr: Union[str, collections.abc.Mapping], expected: Optional[Union[unfurl.result.ResultsList, list, unfurl.result.Result]] = None, schema: Optional[collections.abc.Mapping] = None, name: Optional[str] = None, required: bool = True, wantList: bool = False, target: Optional[unfurl.runtime.EntityInstance] = None) → unfurl.configurator.Dependency
Parameters

  • expr (Union[str, collections.abc.Mapping]) –

  • expected (Optional[Union[unfurl.result.ResultsList, list, unfurl.result.Result]]) –

  • schema (Optional[collections.abc.Mapping]) –

  • name (Optional[str]) –

  • required (bool) –

  • wantList (bool) –

  • target (Optional[unfurl.runtime.EntityInstance]) –

Return type

unfurl.configurator.Dependency

add_message(message: object) → None
Parameters

message (object) –

Return type

None

apply_work_folders(*names: str) → None
Parameters

names (str) –

Return type

None

property connections: unfurl.configurator._ConnectionsMap
create_sub_task(operation: Optional[Union[str, unfurl.planrequests.ConfigurationSpec]] = None, resource: Optional[unfurl.runtime.EntityInstance] = None, inputs: Optional[dict] = None, persist: bool = False, required: Optional[bool] = None) → Optional[unfurl.planrequests.TaskRequest]

Create a subtask that will be executed if yielded by unfurl.configurator.Configurator.run()

Parameters

  • operation (str) – The operation call (like interface.operation)

  • resource (NodeInstance) –

  • inputs (Optional[dict]) –

  • persist (bool) –

  • required (Optional[bool]) –

Returns

TaskRequest

Return type

Optional[unfurl.planrequests.TaskRequest]

discard_work_folders() → None
Return type

None

done(success: Optional[bool] = None, modified: Optional[bool] = None, status: Optional[unfurl.support.Status] = None, result: Optional[Union[dict, str]] = None, outputs: Optional[dict] = None, captureException: Optional[object] = None) → unfurl.configurator.ConfiguratorResult

unfurl.configurator.Configurator.run() should call this method and yield its return value before terminating.

>>> yield task.done(True)
Parameters

  • success (bool) – indicates if this operation completed without an error.

  • modified (bool) – (optional) indicates whether the physical instance was modified by this operation.

  • status (Status) – (optional) should be set if the operation changed the operational status of the target instance. If not specified, the runtime will updated the instance status as needed, based the operation preformed and observed changes to the instance (attributes changed).

  • result (dict) – (optional) A dictionary that will be serialized as YAML into the changelog, can contain any useful data about these operation.

  • outputs (dict) – (optional) Operation outputs, as specified in the topology template.

  • captureException (Optional[object]) –

Returns

ConfiguratorResult

Return type

unfurl.configurator.ConfiguratorResult

property environ: Dict[str, str]
fail_work_folders() → None
Return type

None

find_connection(target: unfurl.runtime.NodeInstance, relation: str = 'tosca.relationships.ConnectsTo') → Optional[unfurl.runtime.RelationshipInstance]

Find a relationship that this task can use to connect to the given instance. First look for relationship between the task’s target instance and the given instance. If none is found, see if there a default connection of the given type.

Parameters

  • target (NodeInstance) – The instance to connect to.

  • relation (str, optional) – The relationship type. Defaults to “tosca.relationships.ConnectsTo”.

Returns

The connection instance.

Return type

RelationshipInstance or None

find_instance(name: str) → unfurl.runtime.NodeInstance
Parameters

name (str) –

Return type

unfurl.runtime.NodeInstance

get_environment(addOnly: bool, env: Optional[dict] = None) → dict

Return a dictionary of environment variables applicable to this task.

Parameters

  • addOnly (bool) – If addOnly is False all variables in the current os environment will be included otherwise only variables added will be included.

  • env (Optional[dict]) –

Returns

dict:

Return type

dict

Variable sources (by order of preference, lowest to highest): 1. The ensemble’s environment 2. Variables set by the connections that are available to this operation. 3. Variables declared in the operation’s environment section.

get_settings() → dict
Return type

dict

get_work_folder(location: Optional[str] = None) → unfurl.projectpaths.WorkFolder
Parameters

location (Optional[str]) –

Return type

unfurl.projectpaths.WorkFolder

property inputs: unfurl.result.ResultsMap

Exposes inputs and task settings as expression variables, so they can be accessed like:

eval: $inputs::param

or in jinja2 templates:

{{ inputs.param }}

query(query: Union[str, dict], dependency: bool = False, name: Optional[str] = None, required: bool = False, wantList: bool = False, resolveExternal: bool = True, strict: bool = True, vars: Optional[dict] = None, throw: bool = False) → Optional[Union[unfurl.result.ResultsList, unfurl.result.Result, List[unfurl.result.Result]]]
Parameters

  • query (Union[str, dict]) –

  • dependency (bool) –

  • name (Optional[str]) –

  • required (bool) –

  • wantList (bool) –

  • resolveExternal (bool) –

  • strict (bool) –

  • vars (Optional[dict]) –

  • throw (bool) –

Return type

Optional[Union[unfurl.result.ResultsList, unfurl.result.Result, List[unfurl.result.Result]]]

remove_dependency(name: str) → Optional[unfurl.configurator.Dependency]
Parameters

name (str) –

Return type

Optional[unfurl.configurator.Dependency]

restore_envvars()
sensitive(value: object) → Union[unfurl.logs.sensitive, object]

Mark the given value as sensitive. Sensitive values will be encrypted or redacted when outputed.

Returns

A copy of the value converted the appropriate subtype of unfurl.logs.sensitive value or the value itself if it can’t be converted.

Return type

sensitive

Parameters

value (object) –

set_envvars()

Update os.environ with the task’s environ and save the current one so it can be restored by restore_envvars

set_work_folder(location: str = 'operation', preserve: Optional[bool] = None, always_apply: bool = False) → unfurl.projectpaths.WorkFolder
Parameters

  • location (str) –

  • preserve (Optional[bool]) –

  • always_apply (bool) –

Return type

unfurl.projectpaths.WorkFolder

update_instances(instances: Union[str, List]) → Tuple[Optional[unfurl.planrequests.JobRequest], List[unfurl.util.UnfurlTaskError]]

Notify Unfurl of new or changes to instances made while the configurator was running.

Operational status indicates if the instance currently exists or not. This will queue a new child job if needed.

To run the job based on the supplied spec immediately, yield the returned JobRequest.

- name:     aNewInstance
  template: aNodeTemplate
  parent:   HOST
  attributes:
     anAttribute: aValue
  readyState:
    local: ok
    state: state
- name:     SELF
  attributes:
      anAttribute: aNewValue
Parameters

instances (list or str) – Either a list or string that is parsed as YAML.

Return type

Tuple[Optional[unfurl.planrequests.JobRequest], List[unfurl.util.UnfurlTaskError]]

property vars: dict

A dictionary of the same variables that are available to expressions when evaluating inputs.

Internal classes supporting the runtime.

class unfurl.support.NodeState(value)

An enumeration.

configured = 5
configuring = 4
created = 3
creating = 2
deleted = 11
deleting = 10
error = 12
initial = 1
started = 7
starting = 6
stopped = 9
stopping = 8
class unfurl.support.Priority(value)

An enumeration.

critical = 3
ignore = 0
optional = 1
required = 2
class unfurl.support.Reason
add = 'add'
check = 'check'
degraded = 'degraded'
error = 'error'
force = 'force'
missing = 'missing'
prune = 'prune'
reconfigure = 'reconfigure'
run = 'run'
update = 'update'
upgrade = 'upgrade'
class unfurl.support.Status(value)

An enumeration.

absent = 5
property color
degraded = 2
error = 3
ok = 1
pending = 4
unknown = 0
class unfurl.result.ChangeRecord(jobId: Optional[str] = None, startTime: Optional[datetime.datetime] = None, taskId: int = 0, previousId: Optional[str] = None, parse: Optional[str] = None)

A ChangeRecord represents a job or task in the change log file. It consists of a change ID and named attributes.

A change ID is an identifier with this sequence of 12 characters: - “A” serves as a format version identifier - 7 alphanumeric characters (0-9, A-Z, and a-z) encoding the date and time the job ran. - 4 hexadecimal digits encoding the task id

Parameters

  • jobId (Optional[str]) –

  • startTime (Optional[datetime.datetime]) –

  • taskId (int) –

  • previousId (Optional[str]) –

  • parse (Optional[str]) –

classmethod format_log(changeId: str, attributes: dict) → str

format: changeidtkey=valuetkey=value

Parameters

  • changeId (str) –

  • attributes (dict) –

Return type

str

log(attributes: Optional[dict] = None) → str

changeidtkey=valuetkey=value

Parameters

attributes (Optional[dict]) –

Return type

str

Runtime module

This module defines the core model and implements the runtime operations of the model.

The state of the system is represented as a collection of Instances. Each instance have a status; attributes that describe its state; and a TOSCA template which describes its capabilities, relationships and available interfaces for configuring and interacting with it.

class unfurl.runtime.Operational

This is an abstract base class for Jobs, Resources, and Configurations all have a Status associated with them and all use the same algorithm to compute their status from their dependent resouces, tasks, and configurations

static aggregate_status(statuses: Iterable[unfurl.runtime.Operational], seen: Dict[int, unfurl.runtime.Operational]) → Optional[unfurl.support.Status]

Returns: ok, degraded, pending or None

If there are no instances, return None If any required are not operational, return pending or error If any other are not operational or degraded, return degraded Otherwise return ok. (Instances with priority set to “ignore” are ignored.)

Parameters
Return type

Optional[unfurl.support.Status]

get_operational_dependencies() → Iterable[unfurl.runtime.Operational]

Return an iterator of Operational object that this instance depends on to be operational.

Return type

Iterable[unfurl.runtime.Operational]

has_changed(changeset) → bool

Whether or not this object changed since the give ChangeRecord.

Return type

bool

class unfurl.runtime.OperationalInstance(status: Optional[Union[unfurl.runtime.OperationalInstance, int, str]] = None, priority: Optional[Union[int, str, unfurl.support.Priority]] = None, manualOveride: Optional[Union[unfurl.runtime.OperationalInstance, int, str]] = None, lastStateChange: Optional[str] = None, lastConfigChange: Optional[str] = None, state: Optional[unfurl.support.NodeState] = None)

A concrete implementation of Operational

Parameters
Return type

None

get_operational_dependencies() → Iterable[unfurl.runtime.Operational]

Return an iterator of Operational object that this instance depends on to be operational.

Return type

Iterable[unfurl.runtime.Operational]

property local_status: Optional[unfurl.support.Status]

The local_status property.

property manual_overide_status: Optional[unfurl.support.Status]

The manualOverideStatus property.

property priority: Optional[unfurl.support.Priority]

The priority property.

property state: Optional[unfurl.support.NodeState]

The state property.

APIs for controlling Unfurl

Localenv module

Classes for managing the local environment.

Repositories can optionally be organized into projects that have a local configuration.

By convention, the “home” project defines a localhost instance and adds it to its context.

class unfurl.localenv.LocalEnv(manifestPath: Optional[str] = None, homePath: Optional[str] = None, parent: Optional[unfurl.localenv.LocalEnv] = None, project: Optional[unfurl.localenv.Project] = None, can_be_empty: bool = False, override_context: Optional[str] = None, overrides: Optional[Dict[str, Any]] = None, readonly: Optional[bool] = False)

This class represents the local environment that an ensemble runs in, including the local project it is part of and the home project.

Parameters

  • manifestPath (str) –

  • homePath (Optional[str]) –

  • parent (Optional[unfurl.localenv.LocalEnv]) –

  • project (Optional[unfurl.localenv.Project]) –

  • can_be_empty (bool) –

  • override_context (Optional[str]) –

  • overrides (Optional[Dict[str, Any]]) –

  • readonly (Optional[bool]) –

Return type

None

find_git_repo(repoURL: str, revision: Optional[str] = None) → Optional[unfurl.repo.GitRepo]
Parameters

  • repoURL (str) –

  • revision (Optional[str]) –

Return type

Optional[unfurl.repo.GitRepo]

find_or_create_working_dir(repoURL: str, revision: Optional[str] = None, basepath: Optional[str] = None, checkout_args: dict = {}) → Tuple[Optional[unfurl.repo.GitRepo], Optional[str], Optional[bool]]
Parameters

  • repoURL (str) –

  • revision (Optional[str]) –

  • basepath (Optional[str]) –

  • checkout_args (dict) –

Return type

Tuple[Optional[unfurl.repo.GitRepo], Optional[str], Optional[bool]]

find_path_in_repos(path: str, importLoader: Optional[Any] = None) → Tuple[Optional[unfurl.repo.GitRepo], Optional[str], Optional[str], Optional[bool]]

If the given path is part of the working directory of a git repository return that repository and a path relative to it

Parameters

  • path (str) –

  • importLoader (Optional[Any]) –

Return type

Tuple[Optional[unfurl.repo.GitRepo], Optional[str], Optional[str], Optional[bool]]

find_project(testPath: str) → Optional[unfurl.localenv.Project]

Walk parents looking for unfurl.yaml

Parameters

testPath (str) –

Return type

Optional[unfurl.localenv.Project]

get_context(context: Optional[dict] = None) → dict

Return a new context that merges the given context with the local context.

Parameters

context (Optional[dict]) –

Return type

dict

get_external_manifest(location: dict, skip_validation: bool, safe_mode: bool) → Optional[YamlManifest]
Parameters

  • location (dict) –

  • skip_validation (bool) –

  • safe_mode (bool) –

Return type

Optional[YamlManifest]

get_local_instance(name: str, context: dict) → Tuple[unfurl.runtime.NodeInstance, dict]
Parameters

  • name (str) –

  • context (dict) –

Return type

Tuple[unfurl.runtime.NodeInstance, dict]

get_manifest(path: Optional[str] = None, skip_validation: bool = False, safe_mode: Optional[bool] = None) → YamlManifest
Parameters

  • path (Optional[str]) –

  • skip_validation (bool) –

  • safe_mode (Optional[bool]) –

Return type

YamlManifest

get_paths() → List[str]

If asdf is installed, build a PATH list from .toolversions found in the current project and the home project.

Return type

List[str]

get_project(path: str, homeProject: Optional[unfurl.localenv.Project]) → unfurl.localenv.Project
Parameters
Return type

unfurl.localenv.Project

get_runtime() → Optional[str]
Return type

Optional[str]

get_vault()
get_vault_password(vaultId: str = 'default') → Optional[str]
Parameters

vaultId (str) –

Return type

Optional[str]

map_value(val: Any, env_rules: Optional[dict]) → Any

Evaluate using project home as a base dir.

Parameters

  • val (Any) –

  • env_rules (Optional[dict]) –

Return type

Any

parent: Optional[unfurl.localenv.LocalEnv] = None
project: Optional[unfurl.localenv.Project] = None
class unfurl.localenv.Project(path: str, homeProject: Optional[unfurl.localenv.Project] = None, overrides: Optional[dict] = None, readonly: Optional[bool] = False)

A Unfurl project is a folder that contains at least a local configuration file (unfurl.yaml), one or more ensemble.yaml files which maybe optionally organized into one or more git repositories.

Parameters

  • path (str) –

  • homeProject (Optional[Project]) –

  • overrides (Optional[dict]) –

  • readonly (Optional[bool]) –

create_working_dir(gitUrl: str, ref: Optional[str] = None) → unfurl.repo.GitRepo
Parameters

  • gitUrl (str) –

  • ref (Optional[str]) –

Return type

unfurl.repo.GitRepo

find_ensemble_by_name(name: str) → Optional[dict]
Parameters

name (str) –

Return type

Optional[dict]

find_ensemble_by_path(path: str) → Optional[dict]
Parameters

path (str) –

Return type

Optional[dict]

find_git_repo(repoURL: str, revision: Optional[str] = None) → Optional[unfurl.repo.GitRepo]
Parameters

  • repoURL (str) –

  • revision (Optional[str]) –

Return type

Optional[unfurl.repo.GitRepo]

find_git_repo_from_repository(repoSpec: toscaparser.repositories.Repository) → Optional[unfurl.repo.Repo]
Parameters

repoSpec (toscaparser.repositories.Repository) –

Return type

Optional[unfurl.repo.Repo]

find_or_clone(repo: unfurl.repo.GitRepo) → unfurl.repo.GitRepo
Parameters

repo (unfurl.repo.GitRepo) –

Return type

unfurl.repo.GitRepo

find_or_create_working_dir(repoURL: str, revision: Optional[str] = None) → Optional[unfurl.repo.GitRepo]
Parameters

  • repoURL (str) –

  • revision (Optional[str]) –

Return type

Optional[unfurl.repo.GitRepo]

static find_path(testPath: str, stopPath: Optional[str] = None) → Optional[str]

Walk parents looking for unfurl.yaml

Parameters

  • testPath (str) –

  • stopPath (Optional[str]) –

Return type

Optional[str]

find_path_in_repos(path: str, importLoader: Optional[Any] = None) → Tuple[Optional[unfurl.repo.GitRepo], Optional[str], Optional[str], Optional[bool]]

If the given path is part of the working directory of a git repository return that repository and a path relative to it

Parameters

  • path (str) –

  • importLoader (Optional[Any]) –

Return type

Tuple[Optional[unfurl.repo.GitRepo], Optional[str], Optional[str], Optional[bool]]

static get_asdf_paths(projectRoot, asdfDataDir, toolVersions={}) → List[str]
Return type

List[str]

get_context(contextName: Optional[str], context: Optional[dict] = None) → dict
Parameters

  • contextName (Optional[str]) –

  • context (Optional[dict]) –

Return type

dict

get_default_context() → Any
Return type

Any

get_default_project_path(context_name: str) → Optional[Any]
Parameters

context_name (str) –

Return type

Optional[Any]

get_managed_project(location: dict, localEnv: unfurl.localenv.LocalEnv) → Optional[unfurl.localenv.Project]
Parameters
Return type

Optional[unfurl.localenv.Project]

get_manifest_path(localEnv: unfurl.localenv.LocalEnv, manifestPath: Optional[str], can_be_empty: bool) → Tuple[Optional[unfurl.localenv.Project], str, str]
Parameters
Return type

Tuple[Optional[unfurl.localenv.Project], str, str]

static get_name_from_dir(projectRoot: str) → str
Parameters

projectRoot (str) –

Return type

str

get_relative_path(path: str) → str
Parameters

path (str) –

Return type

str

get_unique_path(name: str) → str
Parameters

name (str) –

Return type

str

get_vault_password(contextName: Optional[str] = None, vaultId: str = 'default') → Optional[str]
Parameters

  • contextName (Optional[str]) –

  • vaultId (str) –

Return type

Optional[str]

get_vault_passwords(contextName: Optional[str] = None) → Iterable[Tuple[str, Union[str, bytes]]]
Parameters

contextName (Optional[str]) –

Return type

Iterable[Tuple[str, Union[str, bytes]]]

is_path_in_project(path: str) → bool
Parameters

path (str) –

Return type

bool

load_yaml_include(yamlConfig: unfurl.yamlloader.YamlConfig, templatePath: Union[str, dict], baseDir, warnWhenNotFound=False, expanded=None, action=None)

This is called while the YAML config is being loaded. Returns (url or fullpath, parsed yaml)

Parameters

  • yamlConfig (unfurl.yamlloader.YamlConfig) –

  • templatePath (Union[str, dict]) –

make_vault_lib(contextName: Optional[str] = None) → Optional[ansible.parsing.vault.VaultLib]
Parameters

contextName (Optional[str]) –

Return type

Optional[ansible.parsing.vault.VaultLib]

property name: str
static normalize_path(path: str) → str
Parameters

path (str) –

Return type

str

register_ensemble(manifestPath: str, *, project: Optional[unfurl.localenv.Project] = None, managedBy: Optional[unfurl.localenv.Project] = None, context: Optional[str] = None) → None
Parameters
Return type

None

register_project(project, for_context=None, changed=False, save_project=True)
search_for_manifest(can_be_empty: bool) → Optional[str]
Parameters

can_be_empty (bool) –

Return type

Optional[str]

property venv: Optional[str]

Job module

A Job is generated by comparing a list of specs with the last known state of the system. Job runs tasks, each of which has a configuration spec that is executed on the running system Each task tracks and records its modifications to the system’s state

class unfurl.job.ConfigChange(parentJob: Optional[unfurl.job.Job] = None, startTime: Optional[datetime.datetime] = None, status: Optional[Union[unfurl.runtime.OperationalInstance, int, str]] = None, previousId: Optional[str] = None, **kw: Any)

Represents a configuration change made to the system. It has a operating status and a list of dependencies that contribute to its status. There are two kinds of dependencies:

  1. Live resource attributes that the configuration’s inputs depend on.

  2. Other configurations and resources it relies on to function properly.

Parameters

  • parentJob (Optional[Job]) –

  • startTime (Optional[datetime.datetime]) –

  • status (Optional[Union[unfurl.runtime.OperationalInstance, int, str]]) –

  • previousId (Optional[str]) –

  • kw (Any) –

Return type

None

get_operational_dependencies() → Iterable[unfurl.runtime.Operational]

Return an iterator of Operational object that this instance depends on to be operational.

Return type

Iterable[unfurl.runtime.Operational]

class unfurl.job.Job(manifest: YamlManifest, rootResource: unfurl.runtime.TopologyInstance, jobOptions: unfurl.job.JobOptions, previousId: Optional[str] = None)

runs ConfigTasks and child Jobs

Parameters

  • manifest (YamlManifest) –

  • rootResource (unfurl.runtime.TopologyInstance) –

  • jobOptions (unfurl.job.JobOptions) –

  • previousId (Optional[str]) –

Return type

None

can_run_task(task: unfurl.job.ConfigTask) → Tuple[bool, str]

Checked at runtime right before each task is run

  • validate inputs

  • check pre-conditions to see if it can be run

  • check task if it can be run

Parameters

task (unfurl.job.ConfigTask) –

Return type

Tuple[bool, str]

get_operational_dependencies() → Iterable[unfurl.job.ConfigTask]

Return an iterator of Operational object that this instance depends on to be operational.

Return type

Iterable[unfurl.job.ConfigTask]

run_task(task: unfurl.job.ConfigTask, depth: int = 0) → unfurl.job.ConfigTask

During each task run: * Notification of metadata changes that reflect changes made to resources * Notification of add or removing dependency on a resource or properties of a resource * Notification of creation or deletion of a resource * Requests a resource with requested metadata, if it doesn’t exist, a task is run to make it so (e.g. add a dns entry, install a package).

Returns a task.

Parameters

  • task (unfurl.job.ConfigTask) –

  • depth (int) –

Return type

unfurl.job.ConfigTask

should_run_task(task: unfurl.job.ConfigTask) → Tuple[bool, str]

Checked at runtime right before each task is run

Parameters

task (unfurl.job.ConfigTask) –

Return type

Tuple[bool, str]

class unfurl.job.JobOptions(**kw: Any)

Options available to select which tasks are run, e.g. read-only

does the config apply to the operation? is it out of date? is it in a ok state?

Parameters

kw (Any) –

Return type

None

unfurl.job.run_job(manifestPath: Optional[str] = None, _opts: Optional[dict] = None) → unfurl.job.Job

Loads the given Ensemble and creates and runs a job.

Parameters

  • manifestPath (str, optional) – If None, it will look for an ensemble in the current working directory.

  • _opts (dict, optional) – the names of the command line options for creating jobs.

Returns

The job that just ran or None if it couldn’t be created.

Return type

(Job)

Init module

This module implements creating and cloning project and ensembles as well Unfurl runtimes.

unfurl.init.clone(source: str, dest: str, ensemble_name: str = 'ensemble', **options: Any) → str

Clone the source ensemble to dest. If dest isn’t in a project, create one. source can point to an ensemble_template, a service_template, an existing ensemble or a folder containing one of those. If it points to a project its default ensemble will be cloned.

Referenced Repositories will be cloned if a git repository or copied if a regular file folder, If the folders already exist they will be copied to new folder unless the git repositories have the same HEAD. but the local repository names will remain the same.

dest

result

Inside source project

new ensemble

missing or empty folder

clone project, new or cloned ensemble

another project

new or cloned ensemble with source as spec

non-empty folder

error
Parameters

  • source (str) –

  • dest (str) –

  • ensemble_name (str) –

  • options (Any) –

Return type

str

Utility classes and functions

class unfurl.logs.sensitive

Base class for marking a value as sensitive. Depending on the context, sensitive values will either be encrypted or redacted when outputed.

exception unfurl.util.UnfurlError(message: object, saveStack: bool = False, log: bool = False)
Parameters

  • message (object) –

  • saveStack (bool) –

  • log (bool) –

Return type

None

exception unfurl.util.UnfurlTaskError(task: TaskView, message: object, log: int = 40, dependency=None)
Parameters

  • task (TaskView) –

  • message (object) –

  • log (int) –

unfurl.util.filter_env(rules: collections.abc.Mapping, env: Optional[Union[Dict, os._Environ[str]]] = None, addOnly: bool = False, sub: Optional[MutableMapping] = None) → Dict[str, str]

Applies the given list of rules to a dictionary of environment variables and returns a new dictionary.

Parameters

  • rules (dict) – A dictionary of rules for adding, removing and filtering environment variables.

  • env (dict, optional) – The environment to apply the give rules to. If env is None it will be set to os.environ. Defaults to None.

  • addOnly (bool, optional) – If addOnly is False (the default) all variables in env will be included in the returned dict, otherwise only variables added by rules will be included

  • sub (Optional[MutableMapping]) –

Return type

Dict[str, str]

Rules applied in the order they are declared in the rules dictionary. The following examples show the different patterns for the rules:

foo : bar

Add foo=bar

+foo

Copy foo from the current environment

+foo : bar

Copy foo, or add foo=bar if it is not present

+!foo*

Copy all name from the current environment except those matching foo*

-!foo

Remove all names except for foo

^foo : /bar/bin

Treat foo like PATH and prepend /bar/bin:$foo

class unfurl.util.sensitive_bytes

Transparent wrapper class to mark bytes as sensitive

decode(*args: List[object], **kwargs: collections.abc.Mapping) → unfurl.util.sensitive_str

Wrapper method to ensure type conversions maintain sensitive context

Parameters

  • args (List[object]) –

  • kwargs (collections.abc.Mapping) –

Return type

unfurl.util.sensitive_str

class unfurl.util.sensitive_dict

Transparent wrapper class to mark a dict as sensitive

class unfurl.util.sensitive_list(iterable=(), /)

Transparent wrapper class to mark a list as sensitive

class unfurl.util.sensitive_str

Transparent wrapper class to mark a str as sensitive

encode(*args: List[object], **kwargs: collections.abc.Mapping) → unfurl.util.sensitive_bytes

Wrapper method to ensure type conversions maintain sensitive context

Parameters

  • args (List[object]) –

  • kwargs (collections.abc.Mapping) –

Return type

unfurl.util.sensitive_bytes

Eval module

Public Api:

map_value - returns a copy of the given value resolving any embedded queries or template strings

Ref.resolve given an expression, returns a ResultList Ref.resolve_one given an expression, return value, none or a (regular) list Ref.is_ref return true if the given diction looks like a Ref

Internal:

eval_ref() given expression (string or dictionary) return list of Result Expr.resolve() given expression string, return list of Result Results._map_value same as map_value but with lazily evaluation

class unfurl.eval.Ref(exp: Union[str, collections.abc.Mapping], vars: Optional[dict] = None, trace: Optional[int] = None)

A Ref objects describes a path to metadata associated with a resource.

Parameters

  • exp (Union[str, collections.abc.Mapping]) –

  • vars (dict) –

  • trace (Optional[int]) –

Return type

None

resolve(ctx: unfurl.eval.RefContext, wantList: Union[bool, str] = True, strict: Optional[bool] = None) → Optional[Union[unfurl.result.ResultsList, unfurl.result.Result, List[unfurl.result.Result], Any]]

If wantList=True (default) returns a ResultList of matches Note that values in the list can be a list or None If wantList=False return resolve_one semantics If wantList=’result’ return a Result

Parameters
Return type

Optional[Union[unfurl.result.ResultsList, unfurl.result.Result, List[unfurl.result.Result], Any]]

resolve_one(ctx: unfurl.eval.RefContext, strict: Optional[bool] = None) → Optional[Union[unfurl.result.ResultsList, unfurl.result.Result, List[unfurl.result.Result]]]

If no match return None If more than one match return a list of matches Otherwise return the match

Note: If you want to distinguish between None values and no match or between single match that is a list and a list of matches use resolve() which always returns a (possible empty) of matches

Parameters
Return type

Optional[Union[unfurl.result.ResultsList, unfurl.result.Result, List[unfurl.result.Result]]]

class unfurl.eval.RefContext(currentResource: ResourceRef, vars: Optional[dict] = None, wantList: Optional[Union[bool, str]] = False, resolveExternal: bool = False, trace: Optional[int] = None, strict: Optional[bool] = None, task: Optional[TaskView] = None)

The context of the expression being evaluated.

Parameters

  • currentResource (ResourceRef) –

  • vars (Optional[dict]) –

  • wantList (Optional[Union[bool, str]]) –

  • resolveExternal (bool) –

  • trace (Optional[int]) –

  • strict (Optional[bool]) –

  • task (Optional[TaskView]) –

Return type

None

unfurl.eval.eval_ref(val: Union[collections.abc.Mapping, str], ctx: unfurl.eval.RefContext, top: bool = False) → List[unfurl.result.Result]

val is assumed to be an expression, evaluate and return a list of Result

Parameters
Return type

List[unfurl.result.Result]