Errors

lume_services.errors

Classes

EnvironmentNotConfiguredError

EnvironmentNotConfiguredError(
    env_vars, validation_error: ValidationError
)

Bases: Exception

Error for marking unconfigured environment.

Source code in lume_services/errors.py

def __init__(self, env_vars, validation_error: ValidationError):
    self.env = dict(os.environ)
    self.env_vars = []

    for service in env_vars:
        self.env_vars += env_vars[service]

    self.missing_vars = [var for var in self.env_vars if var not in self.env]

    self.message = "%s. Evironment variables not defined: %s"

    super().__init__(
        self.message, str(validation_error), ", ".join(self.missing_vars)
    )

Attributes

env instance-attribute

env = dict(os.environ)

env_vars instance-attribute

env_vars = []

missing_vars instance-attribute

missing_vars = [
    var for var in self.env_vars if var not in self.env
]

message instance-attribute

message = '%s. Evironment variables not defined: %s'

Functions

FlowNotFoundError

FlowNotFoundError(query)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, query):
    self.query = query
    self.message = "Flow not found for query: %s."
    super().__init__(self.message, self.query)

Attributes

query instance-attribute

query = query

message instance-attribute

message = 'Flow not found for query: %s.'

Functions

FlowOfFlowsNotFoundError

FlowOfFlowsNotFoundError(query)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, query):
    self.query = query
    self.message = "Flow not found for query: %s."
    super().__init__(self.message, self.query)

Attributes

query instance-attribute

query = query

message instance-attribute

message = 'Flow not found for query: %s.'

Functions

ProjectNotFoundError

ProjectNotFoundError(query)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, query):
    self.query = query
    self.message = "Project not found for query: %s."
    super().__init__(self.message, self.query)

Attributes

query instance-attribute

query = query

message instance-attribute

message = 'Project not found for query: %s.'

Functions

ModelNotFoundError

ModelNotFoundError(query)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, query):
    self.query = query
    self.message = "Model not found for query: %s."
    super().__init__(self.message, self.query)

Attributes

query instance-attribute

query = query

message instance-attribute

message = 'Model not found for query: %s.'

Functions

DeploymentNotFoundError

DeploymentNotFoundError(query)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, query):
    self.query = query
    self.message = "Deployment not found for query: %s."
    super().__init__(self.message, self.query)

Attributes

query instance-attribute

query = query

message instance-attribute

message = 'Deployment not found for query: %s.'

Functions

ParameterNotInFlowError

ParameterNotInFlowError(
    parameter_name: str, flow_name: str
)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, parameter_name: str, flow_name: str):
    self.flow_name = flow_name
    self.parameter_name = parameter_name
    self.message = "Parameter %s not in flow %s."
    super().__init__(self.message, self.parameter_name, self.flow_name)

Attributes

flow_name instance-attribute

flow_name = flow_name

parameter_name instance-attribute

parameter_name = parameter_name

message instance-attribute

message = 'Parameter %s not in flow %s.'

Functions

ParentFlowNotInFlowsError

ParentFlowNotInFlowsError(flow_name: str, flows: List[str])

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, flow_name: str, flows: List[str]):
    self.flow_name = flow_name
    self.flows = flows
    self.message = "Parent flow %s not in flows: %s"
    super().__init__(self.message, self.flow_name, ", ".join(self.flows))

Attributes

flow_name instance-attribute

flow_name = flow_name

flows instance-attribute

flows = flows

message instance-attribute

message = 'Parent flow %s not in flows: %s'

Functions

TaskNotInFlowError

TaskNotInFlowError(
    flow_name: str, project_name: str, task_name: str
)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, flow_name: str, project_name: str, task_name: str):
    self.flow_name = flow_name
    self.task_name = task_name
    self.project_name = project_name
    self.message = "Task %s not in flow %s."
    super().__init__(
        self.message, self.task_name, self.project_name, self.flow_name
    )

Attributes

flow_name instance-attribute

flow_name = flow_name

task_name instance-attribute

task_name = task_name

project_name instance-attribute

project_name = project_name

message instance-attribute

message = 'Task %s not in flow %s.'

Functions

TaskNotCompletedError

TaskNotCompletedError(
    task_slug: str, flow_id: str, flow_run_id: str
)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, task_slug: str, flow_id: str, flow_run_id: str):
    self.flow_id = flow_id
    self.flow_run_id = flow_run_id
    self.task_slug = task_slug
    self.message = (
        "Task with slug: %s not completed for flow_run_id: %s, flow_id: %s."
    )
    super().__init__(self.message, self.task_slug, self.flow_run_id, self.flow_id)

Attributes

flow_id instance-attribute

flow_id = flow_id

flow_run_id instance-attribute

flow_run_id = flow_run_id

task_slug instance-attribute

task_slug = task_slug

message instance-attribute

message = "Task with slug: %s not completed for flow_run_id: %s, flow_id: %s."

Functions

FlowFailedError

FlowFailedError(
    flow_id: str,
    flow_run_id: str,
    exception_message: str = None,
)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, flow_id: str, flow_run_id: str, exception_message: str = None):
    self.flow_id = flow_id
    self.flow_run_id = flow_run_id
    self.exception_message = exception_message
    self.message = "Flow run: %s failed for flow_id: %s. %s"
    super().__init__(
        self.message, self.flow_run_id, self.flow_id, self.exception_message
    )

Attributes

flow_id instance-attribute

flow_id = flow_id

flow_run_id instance-attribute

flow_run_id = flow_run_id

exception_message instance-attribute

exception_message = exception_message

message instance-attribute

message = 'Flow run: %s failed for flow_id: %s. %s'

Functions

EmptyResultError

EmptyResultError(
    flow_id: str, flow_run_id: str, task_slug: Optional[str]
)

Bases: Exception

Source code in lume_services/errors.py

def __init__(self, flow_id: str, flow_run_id: str, task_slug: Optional[str]):
    self.flow_id = flow_id
    self.flow_run_id = flow_run_id
    self.task_slug = task_slug
    if not self.task_slug:
        self.message = (
            "Task with slug: %s for flow run: %s of flow_id: %s has no result."
        )
        super().__init__(
            self.message, self.task_slug, self.flow_run_id, self.flow_id
        )

    else:
        self.message = "Flow run: %s of flow_id: %s has no result."
        super().__init__(self.message, self.flow_run_id, flow_id)

Attributes

flow_id instance-attribute

flow_id = flow_id

flow_run_id instance-attribute

flow_run_id = flow_run_id

task_slug instance-attribute

task_slug = task_slug

message instance-attribute

message = "Task with slug: %s for flow run: %s of flow_id: %s has no result."

Functions

LocalBackendError

LocalBackendError()

Bases: Exception

LocalBackendError indicates that a server-backend operation has been executed against the LocalBackend. Server-backend operations include flow registration and remote execution.

Source code in lume_services/errors.py

def __init__(self):
    self.message = "Cannot run server-backend operation using LocalBackend."
    super().__init__(self.message)

Attributes

message instance-attribute

message = "Cannot run server-backend operation using LocalBackend."

Functions