Utils

General

estuary.utils.general.get_neo4j_node(resource_name, uid)[source]

Get a Neo4j node based on a label and unique identifier.

Parameters
  • resource_name (str) – a neomodel model label

  • uid (str) – a string of the unique identifier defined in the neomodel model

Returns

a neomodel model object

Raises

ValidationError – if the requested resource doesn’t exist or doesn’t have a UniqueIdProperty

estuary.utils.general.inflate_node(result)[source]

Inflate a Neo4j result to a neomodel model object.

Parameters

result (neo4j.v1.types.Node) – a node from a cypher query result

Returns

a model (EstuaryStructuredNode) object

estuary.utils.general.login_required(f)[source]

Decorate a Flask route to validate a token if authentication is enabled.

Parameters

f (function) – the function to wrap

Returns

the wrapper function

Return type

function

estuary.utils.general.str_to_bool(item)[source]

Convert a string to a boolean.

Parameters

item (str) – string to parse

Returns

a boolean equivalent

Return type

boolean

estuary.utils.general.timestamp_to_date(timestamp)[source]

Convert a string timestamp to a date object.

Parameters

timestamp (str) – a generic or ISO-8601 timestamp

Returns

date object of the timestamp

Return type

datetime.date

Raises

ValueError – if the timestamp is an unsupported or invalid format

estuary.utils.general.timestamp_to_datetime(timestamp)[source]

Convert a string timestamp to a datetime object.

Parameters

timestamp (str) – a generic or ISO-8601 timestamp

Returns

datetime object of the timestamp

Return type

datetime.datetime

Raises

ValueError – if the timestamp is an unsupported or invalid format

Story

class estuary.utils.story.BaseStoryManager[source]

A class containing utility methods to create a story for an artifact.

format_story_results(results, requested_item)[source]

Format story results from Neo4j to the API format.

Parameters
Returns

results in API format

Return type

dict

get_sibling_nodes(siblings_node_label, story_node, count=False)[source]

Return sibling nodes with the label siblings_node_label that are related to story_node.

Parameters
  • siblings_node_label (str) – node label for which the siblings count is to be calculated

  • story_node (EstuaryStructuredNode) – node in the story that has the desired relationships with the siblings (specified with siblings_node_label)

  • count (bool) – determines if only count of sibling nodes should be returned or the nodes themselves

Returns

siblings count of curr_node | sibling nodes

Return type

int | EstuaryStructuredNode

get_sibling_nodes_count(results, reverse=False)[source]

Iterate through the results and yield correlated nodes.

Parameters
  • results (list) – contains inflated results from Neo4j

  • reverse (bool) – determines the direction the story is traversed in (i.e. forward/backward)

Returns

yield the results count (int) received from Neo4j

Return type

generator

static get_siblings_description(story_node_display_name, story_node_story_flow, backward)[source]

Generate a description of the siblings.

Parameters
  • story_node_display_name (string) – the preformatted name to be displayed for the story node

  • story_node_story_flow (dict) – has forward and backward relationships of the story node

  • backward (bool) – determines the relationship direction the story node has with the siblings in the story

Returns

returns the appropriate siblings title

Return type

string

static get_story_manager(item, config, limit=False)[source]

Select which story flow to follow.

Parameters
  • item (node) – a Neo4j node whose story is requested by the user

  • config (flask.config.Config) – flask config

  • limit (bool) – specifies if LIMIT keyword should be added to the created cypher query

Returns

instance of one of the story manager classes

Return type

ModuleStoryManager/ContainerStoryManager

get_story_nodes(item, reverse=False, limit=False)[source]

Create a raw cypher query for story of an artifact and query neo4j with it.

Parameters
  • item (node) – a Neo4j node whose story is requested by the user

  • reverse (bool) – specifies the direction to proceed from current node corresponding to the story_flow

  • limit (bool) – specifies if LIMIT keyword should be added to the created cypher query

Returns

story paths for a particular artifact

Return type

list

get_total_lead_time(results)[source]

Get the total lead time - the time from the start of a story until its current state.

Parameters

results (list) – contains inflated results from Neo4j

Returns

the seconds of total time in the story, or None if sufficient data is not available

Return type

int or None

get_total_processing_time(results)[source]

Get the total time spent processing the story.

Parameters

results (list) – contains inflated results from Neo4j

Returns

the seconds of total time spent processing with a flag for inaccurate calculations

Return type

tuple

get_wait_times(results)[source]

Get the wait time between two artifacts for each pair of them, and the sum of these times.

Parameters

results (list) – contains inflated results from Neo4j

Returns

tuple with list of wait time ints in order of the story (oldest to newest), and a total wait time

Return type

tuple

Raises

RuntimeError – if results has less than 2 elements

set_story_labels(requested_node_label, results, reverse=False)[source]

Replace Neo4j labels with appropriate labels of the story flow.

Parameters
  • requested_node_label (string) – label of the node requested by the user

  • results (list) – nodes in a story/path

  • reverse (bool) – determines if the results are in reverse order of the story flow

Returns

results with story/path labels

Return type

list

abstract story_flow(label)[source]

Get the next/previous node in a story flow/pipeline path.

Parameters

label (str) – Neo4j node label

Returns

uid and relationship information in both forward and backward directions

Return type

dict

class estuary.utils.story.ContainerStoryManager[source]

A class containing utility methods to create a container story.

is_valid()[source]

Determine if the story path matches the returned story.

Returns

whether story is valid for this story path

Return type

bool

story_flow(label)[source]

Get the next/previous node in a story flow/pipeline path.

Parameters

label (str) – Neo4j node label

Returns

uid and relationship information in both forward and backward directions

Return type

dict

class estuary.utils.story.ModuleStoryManager[source]

A class containing utility methods to create a module story.

is_valid()[source]

Determine if the story path matches the returned story.

Returns

whether story is valid for this story path

Return type

bool

story_flow(label)[source]

Get the next/previous node in a story flow/pipeline path.

Parameters

label (str) – Neo4j node label

Returns

uid and relationship information in both forward and backward directions

Return type

dict