gloe.collection

class gloe.collection.Filter(filter_transformer)

Bases: Generic[_T], Transformer[Iterable[_T], list[_T]]

Transformer used to filter values in an iterable using other transformers instead of functions.

Example

In this example, we fetch a list of users and then filter the administrators.:

@transformer
def is_admin(user: User) -> bool: ...

get_admin_users: Transformer[Group, Iterable[User]] = (
    get_users >> Filter(is_admin)
)

Parameters:

filter_transformer (Transformer[_T, bool]) – transformer applied to each item of the input iterable and check if this item must be dropped or not.

transform(data)

Parameters:

data (Iterable[_T]) – incoming iterable to be filtered. The items of this iterable must be of type _T.

Returns:

The filtered iterable.

Return type:

list[_T]

class gloe.collection.FilterAsync(filter_transformer)

Bases: Generic[_T], AsyncTransformer[Iterable[_T], list[_T]]

Async transformer used to filter values in an iterable using other async transformers instead of functions.

Example

In this example, we fetch a list of users and then filter the administrators.:

@async_transformer
async def check_is_admin(user: User) -> bool: ...

get_admin_users: AsyncTransformer[Group, list[User]] = (
    get_users >> FilterAsync(check_is_admin)
)

Parameters:

• filter_transformer (AsyncTransformer[_T, bool]) – async transformer applied to each item of the input iterable

• not. (and check if this item must be dropped or)

async transform_async(data)

Parameters:

data (Iterable[_T]) – incoming iterable to be filtered. The items of this iterable must be of type _T.

Returns:

The filtered iterable.

Return type:

list[_T]

class gloe.collection.Map(mapping_transformer)

Bases: Generic[_T, _U], Transformer[Iterable[_T], list[_U]]

Transformer used to map values in an iterable using other transformers instead of functions.

Example

In this example, we fetch a list of users that belongs to a group. Then, we get the posts of each user and finally turn it into a single list of posts.:

@transformer
def get_user_posts(user: User) -> list[Post]: ...

get_posts_by_group: Transformer[Group, list[Post]] = (
    get_users_by_group >> Map(get_user_posts) >> flatten
)

Parameters:

mapping_transformer (Transformer[_T, _U]) – transformer applied to each item of the input iterable the yield the mapped item of the output iterable.

transform(data)

Parameters:

data (Iterable[_T]) – incoming iterable to be mapped. The items of this iterable must be of type _T.

Returns:

The mapped iterable. The items of this new iterable are of type _U.

Return type:

list[_U]

class gloe.collection.MapAsync(mapping_transformer)

Bases: Generic[_T, _U], AsyncTransformer[Iterable[_T], list[_U]]

Transformer used to map values in an iterable using other async transformers instead of functions.

Example

In this example, we fetch a list of users that belongs to a group. Then, we get the posts of each user and finally turn it into a single list of posts.:

@async_transformer
async def get_user_posts(user: User) -> list[Post]: ...

get_posts_by_group: AsyncTransformer[Group, list[Post]] = (
    get_users_by_group >> MapAsync(get_user_posts) >> flatten
)

Parameters:

mapping_transformer (AsyncTransformer[_T, _U]) – async transformer applied to each item of the input iterable the yield the mapped item of the output iterable.

async transform_async(data)

Parameters:

data (Iterable[_T]) – incoming iterable to be mapped. The items of this iterable must be of type _T.

Returns:

The mapped iterable. The items of this new iterable are of type _U.

Return type:

list[_U]

class gloe.collection.MapOver(iterable, mapping_transformer)

Bases: Generic[_T, _U], Transformer[_T, list[_U]]

Parameters:

• iterable (Iterable[_S])

• mapping_transformer (Transformer[tuple[_T, _S], _U])

transform(data)

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

Parameters:

data (_T)

Return type:

list[_U]

class gloe.collection.MapOverAsync(iterable, mapping_transformer)

Bases: Generic[_T, _U], AsyncTransformer[_T, list[_U]]

Parameters:

• iterable (Iterable[_S])

• mapping_transformer (AsyncTransformer[tuple[_T, _S], _U])

async transform_async(data)

Method to perform the transformation asynchronously.

Parameters:

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

Returns:

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

Return type:

list[_U]