Skip to content

Conduit

Conduits are "job" objects that holds steps and executes them one by one.

If you want to create a new job, just create a new instance of this Conduit object. To create new steps, use create_step() or initialize ConduitStep objects. Lastly, to execute all steps, call the run() coroutine.

Attributes:

Name Type Description
id Optional[str]

An ID that used to identify the job. It doesn't have any effect in code, so it doesn't matter which ID you passed.

name Optional[str]

A name that used to identify the job. It doesn't have any effect in code, so it doesn't matter which name you passed.

steps List[ConduitStep]

List of ConduitStep that added to this job.

tags List[str]

A list of tags that this job has. Tags can be useful to limit the availability to blocks. If you add a tag to a ConduitBlock, only jobs has the tag wil able to use that block. For example, if a block has "A" and "B" tags, then the job must have both "A" and "B" tags too (if job has extra tags such as "C", it doesn't affect the result of the tag checking operation), otherwise using that block will result in a FORBIDDEN_BLOCK status code.

variables Dict[str, ConduitVariable]

A dictionary that contains the job variables. Variables can be set by client or server. They can be accessed inside steps. Note that as these values can be read and edited with users, don't pass any sensitive information here. It is just for creating a temporary storage so user can add and edit any value that they wants.

local_values Dict[str, Any]

A dictionary that contains the job parameters. These local values can be used to store any data that came from an any source. For example if this job has created automatically with one of your app's events, then you can pass the event data here, so users will able to access this value. It is like variables, however local_values is read only to users.

global_values Dict[str, Any]

A dictionary that contains the job globals. Global values work same as variables and local_values, but globals can't be seen by users. Unlike variables and local_values, globals are passed as private function parameters. For example, if a block requires an object as parameter which can't be created by users (For example: You created a block that adds a value to your own app's database, so this operation will require a database instance object to call "write" method on it. But it is impossible to provide a database instance object for user, so need to provide these objects as global value.)

on_step_update Union[Coroutine, Callable, None]

A callable that will be executed when any step in this job has finished (fail or success). It will call the method with two parameters: the Conduit object itself, and step which is updated.

on_job_finish Union[Coroutine, Callable, None]

A callable that will be executed when all steps in this job has finished (fail or success). It will call the method with two parameters: the Conduit object itself, and failed step. If there is no any failed step, then failed step will be None.

step_limit Optional[int]

If you want to add a maximum number to the step count, you can set it here. Users will still able to create steps even if they exceed the limit, however when job has executed, it will exit with an code that says step limit has exceeded. None (which is default) means unlimited count of steps.

blocks List[ConduitBlock]

A list of ConduitBlock objects that will be only available in this job. Make sure to register blocks as "private", otherwise it will have no effect as "public" blocks are already available for all jobs.

block_limit_overrides Dict[str, Optional[int]]

If you want to override "max_uses" count of ConduitBlock objects only for this conduit, you can write block names and set a new limit.

contexts: Dict[str, Any] property readonly

Returns the contexts of the job. Contexts are values that can be seen and used by users. Refer to Context Values section to learn more about contexts.

Returns:

Type Description
Dict[str, Any]

A dictionary that contains information about this job.

success: Optional[bool] property readonly

Returns a boolean about previous execution has exited without any errors.

Returns:

Type Description
Optional[bool]

True if previous execution has exited without any errors. False if previous execution has exited with any errors. None if job has never executed yet.

__init__(self, id=None, name=None, tags=[], variables={}, local_values={}, global_values={}, on_step_update=None, on_job_finish=None, step_limit=None, block_limit_overrides={}, blocks=[]) special

Parameters:

Name Type Description Default
id Optional[str]

An ID that used to identify the job. It doesn't have any effect in code, so it doesn't matter which ID you passed.

None
name Optional[str]

A name that used to identify the job. It doesn't have any effect in code, so it doesn't matter which name you passed.

None
tags List[str]

A list of tags that this job has. Tags can be useful to limit the availability to blocks. If you add a tag to a ConduitBlock, only jobs has the tag wil able to use that block. For example, if a block has "A" and "B" tags, then the job must have both "A" and "B" tags too (if job has extra tags such as "C", it doesn't affect the result of the tag checking operation), otherwise using that block will result in a FORBIDDEN_BLOCK status code.

[]
variables Dict[str, Any]

A dictionary that contains the job variables. Variables can be set by client or server. They can be accessed inside steps. Note that as these values can be read and edited with users, don't pass any sensitive information here. It is just for creating a temporary storage so user can add and edit any value that they wants.

{}
local_values Dict[str, Any]

A dictionary that contains the job parameters. These local values can be used to store any data that came from an any source. For example if this job has created automatically with one of your app's events, then you can pass the event data here, so users will able to access this value. It is like variables, however local_values is read only to users.

{}
global_values Dict[str, Any]

A dictionary that contains the job globals. Global values work same as variables and local_values, but globals can't be seen by users. Unlike variables and local_values, globals are passed as private function parameters. For example, if a block requires an object as parameter which can't be created by users (For example: You created a block that adds a value to your own app's database, so this operation will require a database instance object to call "write" method on it. But it is impossible to provide a database instance object for user, so need to provide these objects as global value.)

{}
on_step_update Union[Coroutine, Callable]

A callable that will be executed when any step in this job has finished (fail or success). It will call the method with two parameters: the Conduit object itself, and step which is updated.

None
on_job_finish Union[Coroutine, Callable]

A callable that will be executed when all steps in this job has finished (fail or success). It will call the method with two parameters: the Conduit object itself, and failed step. If there is no any failed step, then failed step will be None.

None
step_limit Optional[int]

If you want to add a maximum number to the step count, you can set it here. Users will still able to create steps even if they exceed the limit, however when job has executed, it will exit with an code that says step limit has exceeded. None (which is default) means unlimited count of steps.

None
block_limit_overrides Dict[str, Optional[int]]

If you want to override "max_uses" count of ConduitBlock objects only for this conduit, you can write block names and set a new limit. You can also use "*" (wildcard) and "?" (question marks).

{}

create_contexts(self)

Creates and returns a dictionary of contexts. Refer to Context Values section to learn more about contexts.

Note

When running the job, this method will be called before executing every step any will be set as job contexts. If you want to edit the contexts, you can create a custom class that inherits from Conduit class, then override that method.

Returns:

Type Description
Dict[str, Any]

A dictionary that contains information about this job.

create_step(self, action, parameters={}, id=None, forced=False, if_condition=None)

Creates a new step in this job then returns the created step (after adding it to job).

action is the name of the block along with category. For example, if MATH.SUM provided as action, It searches for a block named sum in the Math category.

Refer to ConduitBlock.get about getting a block by its display name.

Parameters:

Name Type Description Default
action str

Name of the block.

required
parameters dict

A dictionary of parameters that will be passed to this block as keyword parameters.

{}
id Optional[str]

An identifier for this step. It needs to be unique among all other steps in the job. If it is None, then the position of the step will be used as step ID.

None
forced bool

Specifies if this step must be executed even if one (or more) of the previous steps fails. Defaults to False.

False
if_condition Union[str, List[str]]

A list or string of if conditions that contains context values. These conditions will be checked before block executing starts, so if one (or more) of conditions fails, then the step will not be executed.

None

Returns:

Type Description
ConduitStep

The created step.

delete_step(self, step, silent=True)

Deletes a step from the job.

If step parameter is int, then it looks for step positions (which starts from 1). If step parameter is str, then it looks for step IDs.

Raises KeyError if step is not found and silent flag is set to False. Otherwise it does nothing when step is not found.

Parameters:

Name Type Description Default
step Union[str, int]

The ID or position of the step that will be deleted.

required
silent bool

If set to True, it won't do anything when step is not found instead of raising an exception.

True

from_dict(data) classmethod

Creates a new Conduit from dictionary.

Parameters:

Name Type Description Default
data Dict[str, Any]

A dictionary that contains configuration for creating new Conduit. Available keys and values are same with __init__ method.

required

Returns:

Type Description
Conduit

Returns the created Conduit.

get_step(self, step, silent=False)

Gets a step in the job.

If step parameter is int, then it looks for step positions (which starts from 1). If step parameter is str, then it looks for step IDs.

Raises KeyError if step is not found and silent flag is set to False. Otherwise it returns None when step is not found.

Parameters:

Name Type Description Default
step Union[str, int]

The ID or position of the step that will be searched for.

required
silent bool

If set to True, it won't do anything when step is not found instead of raising an exception.

False

load_step_list(self, steps)

Creates steps from a list that contains dictionary objects and adds to current job.

Parameters:

Name Type Description Default
steps List[Dict[str, Any]]

List of dictionaries that contains step data. At least the action key is required in all dictionaries but it can include optional keys such as parameters, id, forced and if.

required

run(self) async

Executes all steps in this job one by one. If one of step fails, then next steps will be skipped unless their forced flag is set to True.

Danger

If you use await for job.run(), then this means no other jobs will be executed until current job finishes. If that's the thing that you don't want, then you can use asyncio or alternatives for executing the run() method without await.

update_contexts(self)

Gets the contexts from create_contexts and sets it to job's contexts attribute.

Equivalent to:

job._contexts = job.create_contexts()