Skip to content

ConduitBlock

Represents a function-like object. It holds the parameters, name and the function itself. To access the inner function, use function or __call__ attribute.

Blocks can be defined with @ConduitBlock.make decorator.

@ConduitBlock.make
def say hi():
    return "Hello!"

# Call the function with pydantic validation (if pydantic is available, otherwise it is same as `hi()`):
hi.function() 

# Call the function without pydantic validation:
hi() # Same as hi.__call__()

Attributes:

Name Type Description
function Union[Callable, Coroutine]

The function that wrapped as ConduitBlock. If pydantic has installed, this will be a custom function which comes from pydantic's validate_arguments decorator that enforces parameter types.

name str

A custom name for the block, if it doesn't provided, then it will be the function name. This is always converted to uppercase even if input is lowercase.

category Optional[str]

Name of the ConduitCategory that this block registered in. It will be None if block is not registered in a category.

tags List[str]

List of tags for this block. Tags allows to restrict a block, in order to use a block that has tags, then all block tags need to be exist in Conduit, otherwise using that block will result in a FORBIDDEN_BLOCK status code. For example, if this 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)

private Optional[bool]

If checked to True, then this block won't be registered, so it will not be possible to get this block by its name. Setting it to None will detect if function needs to be private or not by checking if function name starts and ends with double underscore.

max_uses Optional[int]

Limits the usage count for this block per job. For example, if this set to 1, it will not be possible to execute this block more than once. This can be overriden by block_limit_overrides attribute in Conduit

description: Optional[str] inherited property readonly

Returns the docstring of function.

Returns:

Type Description
Optional[str]

Documentation string of the inner function. If there is no documentation string, it will be None.

display_name: str inherited property readonly

Gets the display name for this block. This will be used as identifier when searching for a block.

Returns:

Type Description
str

If block is defined in a ConduitCategory then this will return the category name which is joined with dot (CATEGORY).(BLOCK), otherwise it will just return the block name (BLOCK).

is_coroutine: bool inherited property readonly

Returns True if the inner function is coroutine / async.

Returns:

Type Description
bool

True if the inner function is coroutine (defined with async def syntax), otherwise False.

parameters: Dict[str, inspect.Parameter] inherited property readonly

Returns a dictionary that contains parameters of inner function.

Returns:

Type Description
Dict[str, inspect.Parameter]

A dictionary of function parameters.

raw_function: Union[NoneType, Callable, Coroutine] inherited property readonly

Returns the same raw function that provided when creating the ConduitBlock.

If pydantic is installed, this will return the function without pydantic validation. If pydantic is not installed, this is same as function attribute.

Returns:

Type Description
Union[NoneType, Callable, Coroutine]

Returns the unwrapped raw function.

return_type: Any inherited property readonly

Returns the return type / annotation of the function.

Returns:

Type Description
Any

Any

__init__(self, function, name=None, tags=[], private=None, max_uses=None) special

Creates a new block.

Warning

It is recommended to use @ConduitBlock.make decorator instead of creating a new instance manually.

Parameters:

Name Type Description Default
function Union[Callable, Coroutine]

The function that will be wrapped as ConduitBlock. If pydantic has installed, the function will be wrapped with pydantic's validate_arguments decorator which creates a custom function that enforces parameter types.

required
name str

A custom name for the block, if it doesn't provided, then it uses the function name. This is always converted to uppercase even if input is lowercase.

None
tags List[str]

List of tags for this block. Tags allows to restrict a block, in order to use a block that has tags, then all block tags need to be exist in Conduit, otherwise using that block will result in a FORBIDDEN_BLOCK status code. For example, if this 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)

[]
private Optional[bool]

If checked to True, then this block won't be registered, so it will not be possible to get this block by its name. Setting it to None will detect if function needs to be private or not by checking if function name starts and ends with double underscore.

None
max_uses Optional[int]

Limits the usage count for this block per job. For example, if this set to 1, it will not be possible to execute this block more than once. This can be overriden by block_limit_overrides attribute in Conduit

None

exists_tags(self, tags)

Checks if all block tags are exists in specified tags list.

Returns:

Type Description
bool

True if all block tags are exists in the tags parameter, otherwise False.

get(display_name) classmethod

Gets a block by its display name. You can also search blocks only in a specific category with ConduitCategory.get

Parameters:

Name Type Description Default
display_name str

The display name of the block.

required

Returns:

Type Description
Union[ConduitBlock, ConduitBlock._Partial]

A ConduitBlock object or ConduitPartialBlock if block is not found.

make(function=None, name=None, tags=[], private=None, max_uses=None) staticmethod

Decorate a Callable to ConduitBlock. This registers a function as ConduitBlock.

Given functions will be wrapped with pydantic's validate_arguments decorator (if pydantic has installed in the environment, otherwise it will just wrap without any modification) and returned object will be a ConduitBlock object.

To access the inner function, use function or __call__ attribute.

@ConduitBlock.make
def say hi():
    return "Hello!"

# Call the function with pydantic validation (if pydantic is available, otherwise it is same as `hi()`):
hi.function() 

# Call the function without pydantic validation:
hi() # Same as hi.__call__()

Parameters:

Name Type Description Default
function Union[Callable, Coroutine]

The function that will be wrapped as ConduitBlock. If pydantic has installed, the function will be wrapped with pydantic's validate_arguments decorator which creates a custom function that enforces parameter types.

None
name Optional[str]

A custom name for the block, if it doesn't provided, then it uses the function name. This is always converted to uppercase even if input is lowercase.

None
tags List[str]

List of tags for this block. Tags allows to restrict a block, in order to use a block that has tags, then all block tags need to be exist in Conduit, otherwise using that block will result in a FORBIDDEN_BLOCK status code. For example, if this 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)

[]
private Optional[bool]

If checked to True, then this block won't be registered, so it will not be possible to get this block by its name. Setting it to None will detect if function needs to be private or not by checking if function name starts and ends with double underscore.

None
max_uses Optional[int]

Limits the usage count for this block per job. For example, if this set to 1, it will not be possible to execute this block more than once. This can be overriden by block_limit_overrides attribute in Conduit

None

Returns:

Type Description
ConduitBlock

The created ConduitBlock object.

match_all(pattern) classmethod

Returns a list of ConduitBlocks that matches with the specified wildcard (glob) pattern.

Parameters:

Name Type Description Default
pattern str

The display name of the block.

required

Returns:

Type Description
List[ConduitBlock]

A list of ConduitBlock object or a empty list if no matching block has found.

match_first(pattern) classmethod

Returns the first ConduitBlock that matches with the specified wildcard (glob) pattern.

Parameters:

Name Type Description Default
pattern str

The display name of the block.

required

Returns:

Type Description
Optional[ConduitBlock]

A ConduitBlock object or None if no matching block has found.

prefill_arguments(self, job, **kwargs)

Converts required keyword arguments to positional arguments and returns a list of parameters.

Returns:

Type Description
List[Any]

List of positional arguments that will be passed in to the inner function.