API Reference

Packages

• gloe
• gloe.collection
  • Filter
    • Filter.transform()
  • FilterAsync
    • FilterAsync.transform_async()
  • Map
    • Map.transform()
  • MapAsync
    • MapAsync.transform_async()
  • MapOver
    • MapOver.transform()
  • MapOverAsync
    • MapOverAsync.transform_async()
• gloe.gateways
  • parallel()
  • sequential()
• gloe.utils
  • attach()
  • debug
    • debug.transform()
  • forward
    • forward.transform()
  • forward_incoming()
  • debug
• gloe.experimental
  • EmptyBridgeOnDrop
  • bridge
    • bridge.drop()
    • bridge.pick()

Module contents

Decorators

@gloe.transformer(func: Callable[[A, B, Unpack[Rest]], S]) → MultiArgsTransformer[A, B, Unpack[Rest], S]

@gloe.transformer(func: Callable[[], S]) → Transformer[None, S]

@gloe.transformer(func: Callable[[A], S]) → Transformer[A, S]

Convert a callable to an instance of the Transformer class.

Example

The most common use is as a decorator:

@transformer
def filter_subscribed_users(users: list[User]) -> list[User]:
   ...

subscribed_users = filter_subscribed_users(users_list)

Parameters:

func – A callable with only positional arguments and returns a result. The callable should return an instance of the generic type S specified.

Returns:

An instance of the Transformer class, encapsulating the transformation logic defined in the provided callable.

@gloe.async_transformer(func: Callable[[A, B, Unpack[Rest]], Awaitable[S]]) → MultiArgsAsyncTransformer[A, B, Unpack[Rest], S]

@gloe.async_transformer(func: Callable[[], Awaitable[S]]) → AsyncTransformer[None, S]

@gloe.async_transformer(func: Callable[[A], Awaitable[S]]) → AsyncTransformer[A, S]

Convert a callable to an instance of the AsyncTransformer class.

See also

For more information about this feature, refer to the Async Transformers.

Example

The most common use is as a decorator:

@async_transformer
async def get_user_by_role(role: str) -> list[User]:
   ...

await get_user_by_role("admin")

Parameters:

func – A callable with only positional arguments and returns a coroutine.

Returns:

Returns an instance of the AsyncTransformer class, representing the built async transformer.

@gloe.partial_transformer

This decorator let us create partial transformers, which are transformers that allow for partial application of their arguments. This capability is particularly useful for creating configurable transformer instances where some arguments are preset enhancing modularity and reusability in data processing pipelines.

See also

For further details on partial transformers and their applications, see Partial Transformers.

Example

Here’s how to apply the @partial_transformer decorator to create a transformer with a pre-applied argument:

@partial_transformer
def enrich_data(data: Data, enrichment_type: str) -> Data:
    # Implementation for data enrichment based on the enrichment_type
    ...

# Instantiate a transformer with the 'enrichment_type' pre-set
enrich_with_metadata = enrich_data(enrichment_type="metadata")

# Use the partially applied transformer
get_enriched_data = get_data >> enrich_with_metadata

Parameters:

func (Callable[[Concatenate[A, ~P1]], S]) – A callable with one or more arguments. The first argument is of type A. The subsequent arguments are retained for use during transformer instantiation. This callable returns a value of type S.

Returns:

A callable that receives the same arguments as func, excluding the first one and returns a transformer with incoming type being A and with S as the outcome type.

Return type:

Callable[[~P1], Transformer[A, S]]

@gloe.partial_async_transformer

This decorator enables the creation of partial asynchronous transformers, which are transformers capable of partial argument application. Such functionality is invaluable for crafting reusable asynchronous transformer instances where certain arguments are predetermined, enhancing both modularity and reusability within asynchronous data processing flows.

See also

For additional insights into partial asynchronous transformers and their practical applications, consult Partial Async Transformers.

Example

Utilize the @partial_async_transformer decorator to build a transformer with a pre-set argument:

@partial_async_transformer
async def load_data(user_id: int, data_type: str) -> Data:
    # Logic for loading data based on user_id and data_type
    ...

# Instantiate a transformer with 'data_type' predefined
load_user_data = load_data(data_type="user_profile")

# Subsequent usage requires only the user_id
user_data = await load_user_data(user_id=1234)

Parameters:

func (Callable[[Concatenate[A, ~P1]], Awaitable[S]]) – A callable with one or more arguments, the first of which is of type A. Remaining arguments are preserved for later use during the instantiation of the transformer. This callable must asynchronously return a result of type S, indicating an operation that produces an output of type S upon completion.

Returns:

A callable that receives the same arguments as func, excluding the first one and returns an async transformer with incoming type being A and with S as the outcome type.

Return type:

Callable[[~P1], AsyncTransformer[A, S]]

@gloe.condition

The condition decorator is responsible to build an instance of If[T] class given a Callable[[T], bool].

When instantiating the If class, the parameter name will be the name of the callable.

See also

For additional information about conditions and examples of its usage, consult Conditional Flows.

Example

The below example send a different email for admin and non admin users:

@condition
def is_admin(user: User) -> bool:
    return "admin" in user.roles

send_email = (
    is_admin.Then(fetch_admin_data >> send_admin_email)
    .Else(send_member_email)
)

Parameters:

func (Callable[[In], bool]) – A callable that check somenting about the incoming data and returns a boolean.

Return type:

If[In]

@gloe.ensure(*, incoming: Sequence[Callable[[_T], Any]]) → _ensure_incoming[_T]

@gloe.ensure(*, outcome: Sequence[Callable[[_S], Any]]) → _ensure_outcome[_S]

@gloe.ensure(*, changes: Sequence[Callable[[_T, _S], Any]]) → _ensure_changes[_T, _S]

@gloe.ensure(*, incoming: Sequence[Callable[[_T], Any]], outcome: Sequence[Callable[[_S], Any]]) → _ensure_both[_T, _S]

@gloe.ensure(*, incoming: Sequence[Callable[[_T], Any]], changes: Sequence[Callable[[_T, _S], Any]]) → _ensure_both[_T, _S]

@gloe.ensure(*, outcome: Sequence[Callable[[_T], Any]], changes: Sequence[Callable[[_T, _S], Any]]) → _ensure_both[_T, _S]

@gloe.ensure(*, incoming: Sequence[Callable[[_T], Any]], outcome: Sequence[Callable[[_S], Any]], changes: Sequence[Callable[[_T, _S], Any]]) → _ensure_both[_T, _S]

This decorator is used in transformers to ensure some validation based on its incoming data, outcome data, or both.

These validations are performed by validators. Validators are simple callable functions that validate certain aspects of the input, output, or the differences between them. If the validation fails, it must raise an exception.

The decorator @ensure returns some intermediate classes to assist with the internal logic of Gloe. However, the result of applying it to a transformer is just a new transformer with the exact same attributes, but it includes an additional validation layer.

The motivation of the many overloads is just to allow the user to define different types of validators interchangeably.

See also

For more detailed information about this feature, refer to the Ensurers page.

Parameters:

• incoming (Sequence[Callable[[_T], Any]]) – sequence of validators that will be applied to the incoming data. The type _T refers to the incoming type. Defaut value: [].

• outcome (Sequence[Callable[[_S], Any]]) – sequence of validators that will be applied to the outcome data. The type _S refers to the outcome type. Defaut value: [].

• changes (Sequence[Callable[[_T, _S], Any]]) – sequence of validators that will be applied to both incoming and outcome data. The type _T refers to the incoming type, and type _S refers to the outcome type. Defaut value: [].

Classes

class gloe.BaseTransformer

Bases: Generic[_In, _Out], ABC

property children

Used when a transformer encapsulates other transformer. The encapsulated transformers are called children transformers.

export(path, with_edge_labels=True)

Export Transformer object in dot format

Parameters:

• path (str)

• with_edge_labels (bool)

property label

Label used in visualization.

When the transformer is created by the @transformer decorator, it is the name of the function.

When creating a transformer by extending the Transformer class, it is the name of the class.

property plotting_settings

Defines how the transformer will be plotted.

abstract signature()

Transformer function-like signature

Return type:

Signature

to_dot(path, with_edge_labels=True)

Export Transformer object in dot format

Parameters:

• path (str)

• with_edge_labels (bool)

to_image(path, with_edge_labels=True)

Export Transformer object in a custom image format

Parameters:

• path (str)

• with_edge_labels (bool)

class gloe.Transformer

Bases: BaseTransformer[_I, _O], ABC

A Transformer is the generic block with the responsibility to take an input of type T and transform it to an output of type S.

See also

Read more about this feature in the page Creating a Transformer.

Example

Typical usage example:

class Stringifier(Transformer[dict, str]):
    ...

signature()

Transformer function-like signature

Return type:

Signature

abstract transform(data)

Main method to be implemented and responsible to perform the transformer logic

Parameters:

data (_I)

Return type:

_O

class gloe.AsyncTransformer

Bases: Generic[_In, _Out], BaseTransformer[_In, _Out]

signature()

Transformer function-like signature

Return type:

Signature

abstract async transform_async(data)

Method to perform the transformation asynchronously.

Parameters:

data (_In) – the incoming data passed to the transformer during the pipeline execution.

Returns:

The outcome data, it means, the resulf of the transformation.

Return type:

_Out

class gloe.If(condition, name=None)

Bases: Generic[In]

It is used to start a condition chaining

Example

The below example send a different email for admin and non admin users:

send_email = (
    If(lambda user: "admin" in user.roles)
        .Then(fetch_admin_data >> send_admin_email)
    .Else(send_member_email)
)

Parameters:

• condition (Callable[[In], bool]) – callable returning a boolean.

• name (str | None) – optional argument that adds a label to condition node during plotting.

Exceptions

exception gloe.TransformerException(internal_exception, raiser_transformer, message=None)

Parameters:

• internal_exception (TransformerException | Exception)

• raiser_transformer (BaseTransformer)

• message (str | None)

exception gloe.UnsupportedTransformerArgException(arg)

Parameters:

arg (Any)