| Name | quent JSON |
| Version |
2.1.2
JSON |
| download |
| home_page | |
| Summary | Yet Another Chain Interface. |
| upload_time | 2023-12-05 09:34:00 |
| maintainer | |
| docs_url | None |
| author | |
| requires_python | >=3.7 |
| license | MIT License Copyright (c) 2023 Ohad Drukman Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. |
| keywords |
chain
chaining
cascade
fluent
pipe
piping
interface
syntax
async
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
|
# Quent
#### Yet Another Chain Interface.
## Installation
```
pip install quent
```
## Table of Contents
- [Introduction](#introduction)
- [Real World Example](#real-world-example)
- [Details & Examples](#details--examples)
- [Literal Values](#literal-values)
- [Custom Arguments (`args`, `kwargs`, `Ellipsis`)](#custom-arguments)
- [Flow Modifiers](#flow-modifiers)
- [Chain Template / Reuse](#reusing-a-chain)
- [Chain Nesting](#nesting-a-chain)
- [Pipe Syntax](#pipe-syntax)
- [Safety Callbacks](#safety-callbacks)
- [Comparisons](#comparisons)
- [Iterators](#iterators)
- [Contexts](#contexts)
- [API](#api)
- [Core](#core)
- [`except`, `finally`](#callbacks)
- [Conditionals](#conditionals)
- [Cascade](#cascade)
- [Direct Attribute Access](#direct-attribute-access)
- [Important Notes](#important-notes)
**Suggestions and contributions are more than welcome.**
## Introduction
Quent is an [enhanced](#details--examples), [chain interface](https://en.wikipedia.org/wiki/Method_chaining) implementation for
Python, designed to handle coroutines transparently. The interface and usage of Quent remains exactly the same,
whether you feed it synchronous or asynchronous objects - it can handle almost any use case.
*Every documented API supports both regular functions and coroutines. It will work the exact same way as with a regular
function. Quent automatically awaits any coroutines, even a coroutine that the function passed to `.foreach()` may
return.*
Quent is written in C (using Cython) to minimize it's overhead as much as possible.
As a basic example, take this function:
```python
async def handle_request(id):
data = await fetch_data(id)
data = validate_data(data)
data = normalize_data(data)
return await send_data(data)
```
It uses intermediate variables that only serve to make to code more readable, as opposed to:
```python
async def handle_request(id):
return await send_data(normalize_data(validate_data(await fetch_data(id))))
```
With Quent, we can chain these operations:
```python
from quent import Chain
def handle_request(id):
return Chain(fetch_data, id).then(validate_data).then(normalize_data).then(send_data).run()
```
**Upon evaluation (calling `.run()`), if an awaitable object is detected, Quent wraps it in a Task and returns it.
The task is automatically scheduled for execution and the chain evaluation continues within the task.
As Task objects need not be `await`-ed in order to run, you may or may not `await` it, depending on your needs.**
Besides `Chain`, Quent provides the [Cascade](#cascade) class which implements the [fluent interface](https://en.wikipedia.org/wiki/Fluent_interface).
Quent aims to provide all the necessary tools to handle every use case.
See the full capabilities of Quent in the [API Section](#api).
## Real World Example
This snippet is taken from a thin Redis wrapper I wrote, which supports both the sync and async versions
of `redis` without having a separate implementation for the async version.
```python
def flush(self) -> Any | Coroutine:
""" Execute the current pipeline and return the results, excluding
the results of inline 'expire' commands.
"""
pipe = self.r.pipeline(transaction=self.transaction)
# this applies a bunch of Redis operations onto the `pipe` object.
self.apply_operations(pipe)
return (
Chain(pipe.execute, raise_on_error=True)
.then(self.remove_ignored_commands)
.finally_(pipe.reset, ...)
.run()
)
```
Once the chain runs, it will execute the pipeline commands, remove the unwanted results, and return the rest
of them. Finally, it will reset the `pipe` object. Any function passed to `.finally_()` will **always** be invoked,
even if an exception has been raised during the execution of the chain. The purpose of the `...` here is explained
in the [Ellipsis](#ellipsis) section.
`pipe.execute` and `pipe.reset` are both performing a network request to Redis, and in the case of an `async` Redis
object - are coroutines and would have to be awaited.
Notice that I return without an explicit `await` - if the user of this wrapper has initialized the class with an
async `Redis` instance, they will know that they need to `await` it. This allows me to focus on the actual logic,
without caring about `sync` vs `async`.
Some would say that this pattern can cause unexpected behavior, since it isn't clear when it
will return a Task or not. I see it no differently than any undocumented code - with a proper
and clear documentation (be it an external documentation or just a simple docstring), there shouldn't be
any truly *unexpected* behavior (barring any unknown bugs).
## Details & Examples
### Literal Values
You don't have to pass a callable as a chain item - literal values works just as well.
```python
Chain(fetch_data, id).then(True).run()
```
will execute `fetch_data(id)`, and then return `True`.
### Custom Arguments
You may provide `args` or `kwargs` to a chain item - by doing so, Quent assumes that the item is a callable
and will evaluate it with the provided arguments, instead of evaluating it with the current value.
```python
Chain(fetch_data, id).then(fetch_data, another_id, password=password).run()
```
will execute `fetch_data(id)`, and then `fetch_data(another_id, password=password)`.
#### Ellipsis
The `Ellipsis` / `...` is a special case - if the first argument for *most* functions that register a chain item
or a callback is `...`,
the item will be evaluated without any arguments.
```python
Chain(fetch_data, id).then(do_something, ...).run()
```
will execute `fetch_data(id)`, and then `do_something()`.
### Flow Modifiers
While the default operation of a chain is to, well, chain operations (using `.then()`), there are cases where you may
want to break out of this flow. For this, `Chain` provides the functions `.root()` and `.ignore()`.
They both behave like `.then()`, but with a small difference:
- `.root()` evaluates the item using the root value, instead of the current value.
- `.ignore()` evaluates the item with the current value but will not propagate its result forwards.
There is also a `.root_ignore()` which is the combination of `.root()` and `.ignore()`.
### Reusing A Chain
You may reuse a chain as many times as you wish.
```python
chain = Chain(fetch_data, id).then(validate_data).then(normalize_data).then(send_data)
chain.run()
chain.run()
...
```
There are many cases where you may need to apply the same sequence of operations but with different inputs. Take our
previous example:
```python
Chain(fetch_data, id).then(validate_data).then(normalize_data).then(send_data).run()
```
Instead, we can create a template chain and reuse it, passing different values to `.run()`:
```python
handle_data = Chain().then(validate_data).then(normalize_data).then(send_data)
for id in list_of_ids:
handle_data.run(fetch_data, id)
```
Re-using a `Chain` object will significantly reduce its overhead, as most of the performance hit is due
to the creation of a new `Chain` instance. Nonetheless, the performance hit is negligible and not worth to sacrifice
readability for. So unless it makes sense (or you need to really squeeze out performance),
it's better to create a new `Chain` instance.
### Nesting A Chain
You can nest a `Chain` object within another `Chain` object:
```python
Chain(fetch_data, id)
.then(Chain().then(validate_data).then(normalize_data))
.then(send_data)
.run()
```
A nested chain must always be a template chain (i.e. initialized without arguments).
A nested chain will be evaluated with the current value of the parent chain passed to its `.run()` method.
### Pipe Syntax
Pipe syntax is supported:
```python
from quent import Chain, run
(Chain(fetch_data) | process_data | normalize_data | send_data).run()
Chain(fetch_data) | process_data | normalize_data | send_data | run()
```
You can also use [Pipe](https://github.com/JulienPalard/Pipe) with Quent:
```python
from pipe import where
Chain(get_items).then(where(lambda item: item.is_valid()))
Chain(get_items) | where(lambda item: item.is_valid())
```
### Safety Callbacks
The usage of `except` and `finally` is supported:
```python
Chain(open('<file-path>')).then(do_something_with_file).finally_(lambda f: f.close())
```
Read more about it in [Callbacks](#callbacks).
### Comparisons
Most basic comparison operations are supported:
```python
Chain(get_key).in_(list_of_keys) # == get_key() in list_of_keys
```
See the full list of operations in [Conditionals](#conditionals).
### Iterators
You can easily iterate over the result of something:
```python
Chain(fetch_keys).foreach(do_something_with_key)
```
The full details of `.foreach()` are explained
[here](#foreach).
### Contexts
You can execute a function (or do quite anything you want) inside a context:
```python
Chain(get_lock, id).with_(fetch_data, id)
```
The full details of `.with_()` are explained
[here](#with).
## API
#### Value Evaluation
Most of the methods in the following section receives `value`, `args`, and `kwargs`. Unless explicitly told otherwise,
the evaluation of `value` in all of those methods is roughly equivalent to:
```python
if args[0] is Ellipsis:
return value()
elif args or kwargs:
return value(*args, **kwargs)
elif callable(value):
return value(current_value)
else:
return value
```
The `evaluate_value` function contains the full evaluation logic.
### Core
#### `__init__(value: Any = None, *args, **kwargs)`
Creates a new chain with `value` as the chain's root item. `value` can be anything - a literal value,
a function, a class, etc.
If `args` or `kwargs` are provided, `value` is assumed to be a callable and will be evaluated with those
arguments. Otherwise, a check is performed to determine whether `value` is a callable. If it is, it is
called without any arguments.
Not passing a value will create a template chain (see: [Reusing A Chain](#reusing-a-chain)). You can still normally use
it, but then you must call `.run()` with a value (see the next section).
A few examples:
```python
Chain(42)
Chain(fn, True)
Chain(cls, name='foo')
Chain(lambda v: v*10, 4.2)
```
#### `run(value: Any = None, *args, **kwargs) -> Any | asyncio.Task`
Evaluates the chain and returns the result, or a Task if there are any coroutines in the chain.
If the chain is a template chain (initialized without a value), you must call `.run()` with a value, which will act
as the root item of the chain.
Conversely, if `.run()` is called with a value and the chain is a non-template chain, then an exception will be raised.
The only case where you can both create a template chain and run it without a value is for the `Cascade` class,
which is documented below in [Cascade - Void Mode](#cascade---void-mode).
Similarly to the examples above,
```python
Chain().run(42)
Chain().run(fn, True)
Chain().run(cls, name='foo')
Chain().run(lambda v: v*10, 2)
```
#### `then(value: Any, *args, **kwargs) -> Chain`
Adds `value` to the chain as a chain item. `value` can be anything - a literal value, a function, a class, etc.
Returns the [evaluation](#value-evaluation) of `value`.
This is the main and default way of adding items to the chain.
(see: [Ellipsis](#ellipsis) if you need to invoke `value` without arguments)
```python
Chain(fn).then(False)
Chain(42).then(verify_result)
Chain('<uuid>').then(uuid.UUID)
```
#### `root(value: Any = None, *args, **kwargs) -> Chain`
Like `.then()`, but it first sets the root value as the current value, and then it evaluates `value`
by the default [evaluation procedure](#value-evaluation).
Calling `.root()` without a value simply returns the root value.
Read more in [Flow Modifiers](#flow-modifiers).
```python
Chain(42).then(lambda v: v/10).root(lambda v: v == 42)
```
#### `ignore(value: Any, *args, **kwargs) -> Chain`
Like `.then()`, but keeps the current value unchanged.
In other words, this function does not affect the flow of the chain.
Read more in [Flow Modifiers](#flow-modifiers).
```python
Chain(fetch_data, id).ignore(print).then(validate_data)
```
#### `root_ignore(value: Any, *args, **kwargs) -> Chain`
The combination of `.root()` and `.ignore()`.
```python
Chain(fetch_data, id).then(validate_data).root_ignore(print).then(normalize_data)
```
#### `attr(name: str) -> Chain`
Like `.then()`, but evaluates to `getattr(current_value, name)`.
```python
class A:
@property
def a1(self):
# I return something important
pass
Chain(A()).attr('a1')
ChainAttr(A()).a1
```
#### `attr_fn(name: str, *args, **kwargs) -> Chain`
Like `.attr()`, but evaluates to `getattr(current_value, name)(*args, **kwargs)`.
```python
class A:
def a1(self, foo=None):
# I do something important
pass
Chain(A()).attr_fn('a1', foo=1)
ChainAttr(A()).a1(2)
```
#### Foreach
#### `foreach(fn: Callable) -> Chain`
Iterates over the current value and invokes `fn(element)` for each element. Returns a list
that is the result of `fn(element)` for each `element`.
If the iterator implements `__aiter__`, `async for ...` will be used.
Example:
```python
Chain(list_of_ids).foreach(validate_id).run()
```
will iterate over `list_of_ids`, and return a list that is equivalent to `[validate_id(id) for id in list_of_ids]`.
#### `foreach_do(fn: Callable) -> Chain`
Like `.foreach()`, but returns nothing. In other words, this is the combination of
`.foreach()` and `.ignore()`.
Example:
```python
Chain(list_of_ids)
.foreach_do(Chain().then(fetch_data).then(validate_data).then(normalize_data).then(send_data))
.run()
```
will iterate over `list_of_ids`, invoke the nested chain with each different `id`, and then return `list_of_ids`.
#### With
#### `with_(self, value: Any | Callable = None, *args, **kwargs) -> Chain`
Executes `with current_value as ctx` and evaluates `value` inside the context block,
**with `ctx` as the current value**, and returns the result. If `value` is not provided, returns `ctx`.
This method follows the [default evaluation](#value-evaluation) procedure, so passing `args` or `kwargs`
is perfectly valid.
Depending on `value` (and `args`/`kwargs`), this is roughly equivalent to
```python
with current_value as ctx:
return value(ctx)
```
If the context object implements `__aenter__`, `async with ...` will be used.
Example:
```python
Chain(get_lock, id).with_(fetch_data, id).run()
```
is roughly equivalent to:
```python
with get_lock(id) as lock:
# `lock` is not used here since we passed a custom argument `id`.
return fetch_data(id)
```
#### `with_do(self, value: Any | Callable, *args, **kwargs) -> Chain`
Like `.with_()`, but returns nothing. In other words, this is the combination of
`.with_()` and `.ignore()`.
#### Class Methods
#### `Chain.from_(*args) -> Chain`
Creates a `Chain` template, and registers `args` as chain items.
```python
Chain.from_(validate_data, normalize_data, send_data).run(fetch_data, id)
# is the same as doing
Chain().then(validate_data).then(normalize_data).then(send_data).run(fetch_data, id)
```
### Callbacks
#### `except_(fn: Callable | str, *args, **kwargs) -> Chain`
Register a callback that will be called if an exception is raised anytime during the chain's
evaluation. The callback is evaluated with the root value, or with `args` and `kwargs` if provided.
If `fn` is a string, then it is assumed to be an attribute method of the root value.
```python
Chain(fetch_data).then(validate_data).except_(discard_data)
```
#### `finally_(fn: Callable | str, *args, **kwargs) -> Chain`
Register a callback that will **always** be called after the chain's evaluation. The callback is evaluated with
the root value, or with `args` and `kwargs` if provided.
If `fn` is a string, then it is assumed to be an attribute method of the root value.
```python
Chain(get_id).then(aqcuire_lock).root(fetch_data).finally_(release_lock)
```
### Conditionals
#### `if_(on_true: Any | Callable, *args, **kwargs) -> Chain`
Evaluates the truthiness of the current value (`bool(current_value)`).
If `on_true` is provided and the result is `True`, evaluates `on_true` and returns the result.
If `on_true` is not provided, simply returns the truthiness result (`bool`).
`on_true` may be anything and follows the default [evaluation procedure](#value-evaluation) as described above.
```python
Chain(get_random_number).then(lambda n: n > 5).if_(you_win, prize=1)
```
#### `else_(on_false: Any | Callable, *args, **kwargs) -> Chain`
If a previous conditional result is falsy, evaluates `on_false` and returns the result.
`on_false` may be anything and follows the default [evaluation procedure](#value-evaluation) as described above.
**Can only be called immediately following a conditional.**
```python
Chain(get_random_number).then(lambda n: n > 5).if_(you_win, prize=1).else_(you_lose, cost=10)
```
#### `not_() -> Chain`
- `not current_value`
This method currently does not support the `on_true` argument since it looks confusing.
I might add it in the future.
```python
Chain(is_valid, 'something').not_()
```
#### `eq(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`
- `current_value == value`
```python
Chain(420).then(lambda v: v/10).eq(42)
Chain(420).then(lambda v: v/10).eq(40).else_(on_fail)
```
#### `neq(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`
- `current_value != value`
```python
Chain(420).then(lambda v: v/10).neq(40)
```
#### `is_(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`
- `current_value is value`
```python
Chain(object()).is_(1)
```
#### `is_not(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`
- `current_value is not value`
```python
Chain(object()).is_not(object())
```
#### `in_(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`
- `current_value in value`
```python
Chain('sub').in_('subway')
```
#### `not_in(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`
- `current_value not in value`
```python
Chain('bus').then(lambda s: s[::-1]).not_in('subway')
```
### Cascade
Although considered unpythonic, in some cases the [cascade design](https://en.wikipedia.org/wiki/Fluent_interface)
can be very helpful. The `Cascade` class is identical to `Chain`, except that during the chain's evaluation,
each chain item is evaluated using the root value as an argument
(or in other words, the current value is always the chain's root value).
The return value of `Cascade.run()` is always its root value.
```python
from quent import Cascade
fetched_data = (
Cascade(fetch_data, id)
.then(send_data_to_backup)
.then(lambda data: send_data(data, to_id=1))
.then(print)
.run()
)
```
will execute `fetch_data(id)`, then `send_data_to_backup(data)`, then `send_data(data, to_id=1)`,
and then `print(data)`.
You can also use `Cascade` to make existing classes behave the same way:
```python
from quent import CascadeAttr
class Foo:
def foo(self):
...
async def bar(self):
...
def baz(self):
...
async def get_foo():
f = Foo()
f.foo()
await f.bar()
f.baz()
return f
def better_get_foo():
return CascadeAttr(Foo()).foo().bar().baz().run()
```
`Cascade` works for any kind of object:
```python
from quent import CascadeAttr
CascadeAttr([]).append(1).append(2).append(3).run() == [1, 2, 3]
```
#### Cascade - Void Mode
In some cases it may be desired to run a bunch of independent operations. Using `Cascade`, one can
achieve this by simply not passing a root value to the constructor nor to `.run()`. All the chain items
will not receive any arguments (excluding explicitly provided `args` / `kwargs`).
```python
await (
Cascade()
.then(foo, False)
.then(bar)
.then(baz)
.run()
)
```
will execute `foo(False)`, then `bar()`, then `baz()`.
A void `Cascade` will always return `None`.
### Direct Attribute Access
Both `Chain` and `Cascade` can support "direct" attribute access via the `ChainAttr` and `CascadeAttr` classes.
See the [Cascade](#cascade) section above to see an example of `CascadeAttr` usage. The same principle holds for
`ChainAttr`. Accessing attributes without using the `Attr` subclass is possible using `.attr()` and `.attr_fn()`.
The reason I decided to separate this functionality from the main classes is due to the fact
that it requires overriding `__getattr__`, which drastically increases the overhead of both creating an instance and
accessing any properties / methods. And since I don't think this kind of usage will be common, I decided
to keep this functionality opt-in.
## Important Notes
### Asynchronous `except` and `finally` callbacks
If an except/finally callback is a coroutine function, and an exception is raised *before*
the first coroutine of the chain has been evaluated, or if there aren't any coroutines in the chain - each
callback will be invoked inside a new Task, which you won't have access to.
This is due to the fact that we cannot return the new Task(s) from the `except`/`finally` clauses.
This shouldn't be an issue in most use cases, but important to be aware of.
A `RuntimeWarning` will be emitted in such a case.
As an example, suppose that `fetch_data` is synchronous, and `report_usage` is asynchronous.
```python
Chain(fetch_data).then(raise_exception, ...).finally_(report_usage).run()
```
will execute `fetch_data()`, then `raise_exception()`, and then `report_usage(data)`. But `report_usage(data)` is a
coroutine, and `fetch_data` and `raise_exceptions` are not. Then Quent will wrap `report_usage(data)` in a Task
and "forget" about it.
If you must, you can "force" an async chain by giving it a dummy coroutine:
```python
async def fn(v):
return v
await Chain(fetch_data).then(fn).then(raise_exception, ...).finally_(report_usage).run()
```
This will ensure that `report_usage()` will be awaited properly.
Raw data
{
"_id": null,
"home_page": "",
"name": "quent",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.7",
"maintainer_email": "",
"keywords": "chain,chaining,cascade,fluent,pipe,piping,interface,syntax,async",
"author": "",
"author_email": "Ohad Drukman <drukmano@icloud.com>",
"download_url": "https://files.pythonhosted.org/packages/58/87/d15aaec216ab3f007baf1e741c59cf2809fe9921cef21d5633f38a6cdb52/quent-2.1.2.tar.gz",
"platform": null,
"description": "# Quent\n#### Yet Another Chain Interface.\n\n\n## Installation\n```\npip install quent\n```\n\n## Table of Contents\n- [Introduction](#introduction)\n- [Real World Example](#real-world-example)\n- [Details & Examples](#details--examples)\n - [Literal Values](#literal-values)\n - [Custom Arguments (`args`, `kwargs`, `Ellipsis`)](#custom-arguments)\n - [Flow Modifiers](#flow-modifiers)\n - [Chain Template / Reuse](#reusing-a-chain)\n - [Chain Nesting](#nesting-a-chain)\n - [Pipe Syntax](#pipe-syntax)\n - [Safety Callbacks](#safety-callbacks)\n - [Comparisons](#comparisons)\n - [Iterators](#iterators)\n - [Contexts](#contexts)\n- [API](#api)\n - [Core](#core)\n - [`except`, `finally`](#callbacks)\n - [Conditionals](#conditionals)\n- [Cascade](#cascade)\n- [Direct Attribute Access](#direct-attribute-access)\n- [Important Notes](#important-notes)\n\n**Suggestions and contributions are more than welcome.**\n\n## Introduction\n\nQuent is an [enhanced](#details--examples), [chain interface](https://en.wikipedia.org/wiki/Method_chaining) implementation for\nPython, designed to handle coroutines transparently. The interface and usage of Quent remains exactly the same,\nwhether you feed it synchronous or asynchronous objects - it can handle almost any use case.\n\n*Every documented API supports both regular functions and coroutines. It will work the exact same way as with a regular\nfunction. Quent automatically awaits any coroutines, even a coroutine that the function passed to `.foreach()` may\nreturn.*\n\nQuent is written in C (using Cython) to minimize it's overhead as much as possible.\n\nAs a basic example, take this function:\n```python\nasync def handle_request(id):\n data = await fetch_data(id)\n data = validate_data(data)\n data = normalize_data(data)\n return await send_data(data)\n```\nIt uses intermediate variables that only serve to make to code more readable, as opposed to:\n```python\nasync def handle_request(id):\n return await send_data(normalize_data(validate_data(await fetch_data(id))))\n```\n\nWith Quent, we can chain these operations:\n```python\nfrom quent import Chain\n\ndef handle_request(id):\n return Chain(fetch_data, id).then(validate_data).then(normalize_data).then(send_data).run()\n```\n\n**Upon evaluation (calling `.run()`), if an awaitable object is detected, Quent wraps it in a Task and returns it.\nThe task is automatically scheduled for execution and the chain evaluation continues within the task.\nAs Task objects need not be `await`-ed in order to run, you may or may not `await` it, depending on your needs.**\n\nBesides `Chain`, Quent provides the [Cascade](#cascade) class which implements the [fluent interface](https://en.wikipedia.org/wiki/Fluent_interface).\n\nQuent aims to provide all the necessary tools to handle every use case.\nSee the full capabilities of Quent in the [API Section](#api).\n\n## Real World Example\nThis snippet is taken from a thin Redis wrapper I wrote, which supports both the sync and async versions\nof `redis` without having a separate implementation for the async version.\n```python\ndef flush(self) -> Any | Coroutine:\n \"\"\" Execute the current pipeline and return the results, excluding\n the results of inline 'expire' commands.\n \"\"\"\n pipe = self.r.pipeline(transaction=self.transaction)\n # this applies a bunch of Redis operations onto the `pipe` object.\n self.apply_operations(pipe)\n return (\n Chain(pipe.execute, raise_on_error=True)\n .then(self.remove_ignored_commands)\n .finally_(pipe.reset, ...)\n .run()\n )\n```\nOnce the chain runs, it will execute the pipeline commands, remove the unwanted results, and return the rest\nof them. Finally, it will reset the `pipe` object. Any function passed to `.finally_()` will **always** be invoked,\neven if an exception has been raised during the execution of the chain. The purpose of the `...` here is explained\nin the [Ellipsis](#ellipsis) section.\n\n`pipe.execute` and `pipe.reset` are both performing a network request to Redis, and in the case of an `async` Redis\nobject - are coroutines and would have to be awaited.\nNotice that I return without an explicit `await` - if the user of this wrapper has initialized the class with an\nasync `Redis` instance, they will know that they need to `await` it. This allows me to focus on the actual logic,\nwithout caring about `sync` vs `async`.\n\nSome would say that this pattern can cause unexpected behavior, since it isn't clear when it\nwill return a Task or not. I see it no differently than any undocumented code - with a proper\nand clear documentation (be it an external documentation or just a simple docstring), there shouldn't be\nany truly *unexpected* behavior (barring any unknown bugs).\n\n## Details & Examples\n### Literal Values\nYou don't have to pass a callable as a chain item - literal values works just as well.\n```python\nChain(fetch_data, id).then(True).run()\n```\nwill execute `fetch_data(id)`, and then return `True`.\n\n### Custom Arguments\nYou may provide `args` or `kwargs` to a chain item - by doing so, Quent assumes that the item is a callable\nand will evaluate it with the provided arguments, instead of evaluating it with the current value.\n```python\nChain(fetch_data, id).then(fetch_data, another_id, password=password).run()\n```\nwill execute `fetch_data(id)`, and then `fetch_data(another_id, password=password)`.\n#### Ellipsis\nThe `Ellipsis` / `...` is a special case - if the first argument for *most* functions that register a chain item\nor a callback is `...`,\nthe item will be evaluated without any arguments.\n```python\nChain(fetch_data, id).then(do_something, ...).run()\n```\nwill execute `fetch_data(id)`, and then `do_something()`.\n\n### Flow Modifiers\nWhile the default operation of a chain is to, well, chain operations (using `.then()`), there are cases where you may\nwant to break out of this flow. For this, `Chain` provides the functions `.root()` and `.ignore()`.\nThey both behave like `.then()`, but with a small difference:\n\n- `.root()` evaluates the item using the root value, instead of the current value.\n- `.ignore()` evaluates the item with the current value but will not propagate its result forwards.\n\nThere is also a `.root_ignore()` which is the combination of `.root()` and `.ignore()`.\n\n### Reusing A Chain\nYou may reuse a chain as many times as you wish.\n```python\nchain = Chain(fetch_data, id).then(validate_data).then(normalize_data).then(send_data)\nchain.run()\nchain.run()\n...\n```\n\nThere are many cases where you may need to apply the same sequence of operations but with different inputs. Take our\nprevious example:\n```python\nChain(fetch_data, id).then(validate_data).then(normalize_data).then(send_data).run()\n```\nInstead, we can create a template chain and reuse it, passing different values to `.run()`:\n```python\nhandle_data = Chain().then(validate_data).then(normalize_data).then(send_data)\n\nfor id in list_of_ids:\n handle_data.run(fetch_data, id)\n```\nRe-using a `Chain` object will significantly reduce its overhead, as most of the performance hit is due\nto the creation of a new `Chain` instance. Nonetheless, the performance hit is negligible and not worth to sacrifice\nreadability for. So unless it makes sense (or you need to really squeeze out performance),\nit's better to create a new `Chain` instance.\n\n### Nesting A Chain\nYou can nest a `Chain` object within another `Chain` object:\n```python\nChain(fetch_data, id)\n.then(Chain().then(validate_data).then(normalize_data))\n.then(send_data)\n.run()\n```\nA nested chain must always be a template chain (i.e. initialized without arguments).\n\nA nested chain will be evaluated with the current value of the parent chain passed to its `.run()` method.\n\n### Pipe Syntax\nPipe syntax is supported:\n```python\nfrom quent import Chain, run\n\n(Chain(fetch_data) | process_data | normalize_data | send_data).run()\nChain(fetch_data) | process_data | normalize_data | send_data | run()\n```\nYou can also use [Pipe](https://github.com/JulienPalard/Pipe) with Quent:\n```python\nfrom pipe import where\n\nChain(get_items).then(where(lambda item: item.is_valid()))\nChain(get_items) | where(lambda item: item.is_valid())\n```\n\n### Safety Callbacks\nThe usage of `except` and `finally` is supported:\n```python\nChain(open('<file-path>')).then(do_something_with_file).finally_(lambda f: f.close())\n```\nRead more about it in [Callbacks](#callbacks).\n\n### Comparisons\nMost basic comparison operations are supported:\n```python\nChain(get_key).in_(list_of_keys) # == get_key() in list_of_keys\n```\nSee the full list of operations in [Conditionals](#conditionals).\n\n### Iterators\nYou can easily iterate over the result of something:\n```python\nChain(fetch_keys).foreach(do_something_with_key)\n```\nThe full details of `.foreach()` are explained\n[here](#foreach).\n\n### Contexts\nYou can execute a function (or do quite anything you want) inside a context:\n```python\nChain(get_lock, id).with_(fetch_data, id)\n```\nThe full details of `.with_()` are explained\n[here](#with).\n\n## API\n#### Value Evaluation\nMost of the methods in the following section receives `value`, `args`, and `kwargs`. Unless explicitly told otherwise,\nthe evaluation of `value` in all of those methods is roughly equivalent to:\n```python\nif args[0] is Ellipsis:\n return value()\n\nelif args or kwargs:\n return value(*args, **kwargs)\n\nelif callable(value):\n return value(current_value)\n\nelse:\n return value\n```\nThe `evaluate_value` function contains the full evaluation logic.\n\n### Core\n#### `__init__(value: Any = None, *args, **kwargs)`\nCreates a new chain with `value` as the chain's root item. `value` can be anything - a literal value,\na function, a class, etc.\nIf `args` or `kwargs` are provided, `value` is assumed to be a callable and will be evaluated with those\narguments. Otherwise, a check is performed to determine whether `value` is a callable. If it is, it is\ncalled without any arguments.\n\nNot passing a value will create a template chain (see: [Reusing A Chain](#reusing-a-chain)). You can still normally use\nit, but then you must call `.run()` with a value (see the next section).\n\nA few examples:\n```python\nChain(42)\nChain(fn, True)\nChain(cls, name='foo')\nChain(lambda v: v*10, 4.2)\n```\n\n#### `run(value: Any = None, *args, **kwargs) -> Any | asyncio.Task`\nEvaluates the chain and returns the result, or a Task if there are any coroutines in the chain.\n\nIf the chain is a template chain (initialized without a value), you must call `.run()` with a value, which will act\nas the root item of the chain.\n\nConversely, if `.run()` is called with a value and the chain is a non-template chain, then an exception will be raised.\nThe only case where you can both create a template chain and run it without a value is for the `Cascade` class,\nwhich is documented below in [Cascade - Void Mode](#cascade---void-mode).\n\nSimilarly to the examples above,\n```python\nChain().run(42)\nChain().run(fn, True)\nChain().run(cls, name='foo')\nChain().run(lambda v: v*10, 2)\n```\n\n#### `then(value: Any, *args, **kwargs) -> Chain`\nAdds `value` to the chain as a chain item. `value` can be anything - a literal value, a function, a class, etc.\n\nReturns the [evaluation](#value-evaluation) of `value`.\n\nThis is the main and default way of adding items to the chain.\n\n(see: [Ellipsis](#ellipsis) if you need to invoke `value` without arguments)\n\n```python\nChain(fn).then(False)\nChain(42).then(verify_result)\nChain('<uuid>').then(uuid.UUID)\n```\n\n#### `root(value: Any = None, *args, **kwargs) -> Chain`\nLike `.then()`, but it first sets the root value as the current value, and then it evaluates `value`\nby the default [evaluation procedure](#value-evaluation).\n\nCalling `.root()` without a value simply returns the root value.\n\nRead more in [Flow Modifiers](#flow-modifiers).\n\n```python\nChain(42).then(lambda v: v/10).root(lambda v: v == 42)\n```\n\n#### `ignore(value: Any, *args, **kwargs) -> Chain`\nLike `.then()`, but keeps the current value unchanged.\n\nIn other words, this function does not affect the flow of the chain.\n\nRead more in [Flow Modifiers](#flow-modifiers).\n\n```python\nChain(fetch_data, id).ignore(print).then(validate_data)\n```\n\n#### `root_ignore(value: Any, *args, **kwargs) -> Chain`\nThe combination of `.root()` and `.ignore()`.\n\n```python\nChain(fetch_data, id).then(validate_data).root_ignore(print).then(normalize_data)\n```\n\n#### `attr(name: str) -> Chain`\nLike `.then()`, but evaluates to `getattr(current_value, name)`.\n\n```python\nclass A:\n @property\n def a1(self):\n # I return something important\n pass\n\nChain(A()).attr('a1')\nChainAttr(A()).a1\n```\n\n#### `attr_fn(name: str, *args, **kwargs) -> Chain`\nLike `.attr()`, but evaluates to `getattr(current_value, name)(*args, **kwargs)`.\n\n```python\nclass A:\n def a1(self, foo=None):\n # I do something important\n pass\n\nChain(A()).attr_fn('a1', foo=1)\nChainAttr(A()).a1(2)\n```\n\n#### Foreach\n#### `foreach(fn: Callable) -> Chain`\nIterates over the current value and invokes `fn(element)` for each element. Returns a list\nthat is the result of `fn(element)` for each `element`.\n\nIf the iterator implements `__aiter__`, `async for ...` will be used.\n\nExample:\n```python\nChain(list_of_ids).foreach(validate_id).run()\n```\nwill iterate over `list_of_ids`, and return a list that is equivalent to `[validate_id(id) for id in list_of_ids]`.\n\n#### `foreach_do(fn: Callable) -> Chain`\nLike `.foreach()`, but returns nothing. In other words, this is the combination of\n`.foreach()` and `.ignore()`.\n\nExample:\n```python\nChain(list_of_ids)\n.foreach_do(Chain().then(fetch_data).then(validate_data).then(normalize_data).then(send_data))\n.run()\n```\nwill iterate over `list_of_ids`, invoke the nested chain with each different `id`, and then return `list_of_ids`.\n\n#### With\n#### `with_(self, value: Any | Callable = None, *args, **kwargs) -> Chain`\nExecutes `with current_value as ctx` and evaluates `value` inside the context block,\n**with `ctx` as the current value**, and returns the result. If `value` is not provided, returns `ctx`.\nThis method follows the [default evaluation](#value-evaluation) procedure, so passing `args` or `kwargs`\nis perfectly valid.\n\nDepending on `value` (and `args`/`kwargs`), this is roughly equivalent to\n```python\nwith current_value as ctx:\n return value(ctx)\n```\nIf the context object implements `__aenter__`, `async with ...` will be used.\n\nExample:\n```python\nChain(get_lock, id).with_(fetch_data, id).run()\n```\nis roughly equivalent to:\n```python\nwith get_lock(id) as lock:\n # `lock` is not used here since we passed a custom argument `id`.\n return fetch_data(id)\n```\n\n#### `with_do(self, value: Any | Callable, *args, **kwargs) -> Chain`\nLike `.with_()`, but returns nothing. In other words, this is the combination of\n`.with_()` and `.ignore()`.\n\n#### Class Methods\n#### `Chain.from_(*args) -> Chain`\nCreates a `Chain` template, and registers `args` as chain items.\n\n```python\nChain.from_(validate_data, normalize_data, send_data).run(fetch_data, id)\n# is the same as doing\nChain().then(validate_data).then(normalize_data).then(send_data).run(fetch_data, id)\n```\n\n### Callbacks\n#### `except_(fn: Callable | str, *args, **kwargs) -> Chain`\nRegister a callback that will be called if an exception is raised anytime during the chain's\nevaluation. The callback is evaluated with the root value, or with `args` and `kwargs` if provided.\n\nIf `fn` is a string, then it is assumed to be an attribute method of the root value.\n\n```python\nChain(fetch_data).then(validate_data).except_(discard_data)\n```\n\n#### `finally_(fn: Callable | str, *args, **kwargs) -> Chain`\n\nRegister a callback that will **always** be called after the chain's evaluation. The callback is evaluated with\nthe root value, or with `args` and `kwargs` if provided.\n\nIf `fn` is a string, then it is assumed to be an attribute method of the root value.\n\n```python\nChain(get_id).then(aqcuire_lock).root(fetch_data).finally_(release_lock)\n```\n\n### Conditionals\n#### `if_(on_true: Any | Callable, *args, **kwargs) -> Chain`\nEvaluates the truthiness of the current value (`bool(current_value)`).\nIf `on_true` is provided and the result is `True`, evaluates `on_true` and returns the result.\nIf `on_true` is not provided, simply returns the truthiness result (`bool`).\n\n`on_true` may be anything and follows the default [evaluation procedure](#value-evaluation) as described above.\n\n```python\nChain(get_random_number).then(lambda n: n > 5).if_(you_win, prize=1)\n```\n\n#### `else_(on_false: Any | Callable, *args, **kwargs) -> Chain`\nIf a previous conditional result is falsy, evaluates `on_false` and returns the result.\n\n`on_false` may be anything and follows the default [evaluation procedure](#value-evaluation) as described above.\n\n**Can only be called immediately following a conditional.**\n\n```python\nChain(get_random_number).then(lambda n: n > 5).if_(you_win, prize=1).else_(you_lose, cost=10)\n```\n\n#### `not_() -> Chain`\n- `not current_value`\n\nThis method currently does not support the `on_true` argument since it looks confusing.\nI might add it in the future.\n\n```python\nChain(is_valid, 'something').not_()\n```\n\n#### `eq(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`\n- `current_value == value`\n\n```python\nChain(420).then(lambda v: v/10).eq(42)\nChain(420).then(lambda v: v/10).eq(40).else_(on_fail)\n```\n\n#### `neq(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`\n- `current_value != value`\n\n```python\nChain(420).then(lambda v: v/10).neq(40)\n```\n\n#### `is_(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`\n- `current_value is value`\n\n```python\nChain(object()).is_(1)\n```\n\n#### `is_not(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`\n- `current_value is not value`\n\n```python\nChain(object()).is_not(object())\n```\n\n#### `in_(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`\n- `current_value in value`\n\n```python\nChain('sub').in_('subway')\n```\n\n#### `not_in(value: Any, on_true: Any | Callable = None, *args, **kwargs) -> Chain`\n- `current_value not in value`\n\n```python\nChain('bus').then(lambda s: s[::-1]).not_in('subway')\n```\n\n### Cascade\nAlthough considered unpythonic, in some cases the [cascade design](https://en.wikipedia.org/wiki/Fluent_interface)\ncan be very helpful. The `Cascade` class is identical to `Chain`, except that during the chain's evaluation,\neach chain item is evaluated using the root value as an argument\n(or in other words, the current value is always the chain's root value).\nThe return value of `Cascade.run()` is always its root value.\n```python\nfrom quent import Cascade\n\nfetched_data = (\n Cascade(fetch_data, id)\n .then(send_data_to_backup)\n .then(lambda data: send_data(data, to_id=1))\n .then(print)\n .run()\n)\n```\nwill execute `fetch_data(id)`, then `send_data_to_backup(data)`, then `send_data(data, to_id=1)`,\nand then `print(data)`.\n\nYou can also use `Cascade` to make existing classes behave the same way:\n```python\nfrom quent import CascadeAttr\n\n\nclass Foo:\n def foo(self):\n ...\n async def bar(self):\n ...\n def baz(self):\n ...\n\nasync def get_foo():\n f = Foo()\n f.foo()\n await f.bar()\n f.baz()\n return f\n\ndef better_get_foo():\n return CascadeAttr(Foo()).foo().bar().baz().run()\n```\n\n`Cascade` works for any kind of object:\n```python\nfrom quent import CascadeAttr\n\nCascadeAttr([]).append(1).append(2).append(3).run() == [1, 2, 3]\n```\n\n#### Cascade - Void Mode\nIn some cases it may be desired to run a bunch of independent operations. Using `Cascade`, one can\nachieve this by simply not passing a root value to the constructor nor to `.run()`. All the chain items\nwill not receive any arguments (excluding explicitly provided `args` / `kwargs`).\n```python\nawait (\n Cascade()\n .then(foo, False)\n .then(bar)\n .then(baz)\n .run()\n)\n```\nwill execute `foo(False)`, then `bar()`, then `baz()`.\n\nA void `Cascade` will always return `None`.\n\n### Direct Attribute Access\nBoth `Chain` and `Cascade` can support \"direct\" attribute access via the `ChainAttr` and `CascadeAttr` classes.\nSee the [Cascade](#cascade) section above to see an example of `CascadeAttr` usage. The same principle holds for\n`ChainAttr`. Accessing attributes without using the `Attr` subclass is possible using `.attr()` and `.attr_fn()`.\n\nThe reason I decided to separate this functionality from the main classes is due to the fact\nthat it requires overriding `__getattr__`, which drastically increases the overhead of both creating an instance and\naccessing any properties / methods. And since I don't think this kind of usage will be common, I decided\nto keep this functionality opt-in.\n\n## Important Notes\n### Asynchronous `except` and `finally` callbacks\nIf an except/finally callback is a coroutine function, and an exception is raised *before*\nthe first coroutine of the chain has been evaluated, or if there aren't any coroutines in the chain - each\ncallback will be invoked inside a new Task, which you won't have access to.\nThis is due to the fact that we cannot return the new Task(s) from the `except`/`finally` clauses.\n\nThis shouldn't be an issue in most use cases, but important to be aware of.\nA `RuntimeWarning` will be emitted in such a case.\n\nAs an example, suppose that `fetch_data` is synchronous, and `report_usage` is asynchronous.\n```python\nChain(fetch_data).then(raise_exception, ...).finally_(report_usage).run()\n```\nwill execute `fetch_data()`, then `raise_exception()`, and then `report_usage(data)`. But `report_usage(data)` is a\ncoroutine, and `fetch_data` and `raise_exceptions` are not. Then Quent will wrap `report_usage(data)` in a Task\nand \"forget\" about it.\n\nIf you must, you can \"force\" an async chain by giving it a dummy coroutine:\n```python\nasync def fn(v):\n return v\n\nawait Chain(fetch_data).then(fn).then(raise_exception, ...).finally_(report_usage).run()\n```\nThis will ensure that `report_usage()` will be awaited properly.\n",
"bugtrack_url": null,
"license": "MIT License Copyright (c) 2023 Ohad Drukman Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the \"Software\"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ",
"summary": "Yet Another Chain Interface.",
"version": "2.1.2",
"project_urls": {
"homepage": "https://github.com/drukmano/quent.git"
},
"split_keywords": [
"chain",
"chaining",
"cascade",
"fluent",
"pipe",
"piping",
"interface",
"syntax",
"async"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "5887d15aaec216ab3f007baf1e741c59cf2809fe9921cef21d5633f38a6cdb52",
"md5": "ce99f7844d0b1cbe3647ac6dc1a20683",
"sha256": "6b87a3d0752d349c7e45468412b85d238f09428521647af55e9dceb5c7043224"
},
"downloads": -1,
"filename": "quent-2.1.2.tar.gz",
"has_sig": false,
"md5_digest": "ce99f7844d0b1cbe3647ac6dc1a20683",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.7",
"size": 359817,
"upload_time": "2023-12-05T09:34:00",
"upload_time_iso_8601": "2023-12-05T09:34:00.601403Z",
"url": "https://files.pythonhosted.org/packages/58/87/d15aaec216ab3f007baf1e741c59cf2809fe9921cef21d5633f38a6cdb52/quent-2.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-12-05 09:34:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "drukmano",
"github_project": "quent",
"travis_ci": false,
"coveralls": true,
"github_actions": false,
"lcname": "quent"
}
