# Dev4py-utils
A set of Python regularly used classes/functions
[](https://github.com/dev4py/dev4py-utils/actions/workflows/ci.yml) \
[](https://github.com/dev4py/dev4py-utils/actions/workflows/on_release.yml) \
[](https://github.com/dev4py/dev4py-utils/actions/workflows/weekly_checks.yml) \
[](https://www.python.org/) \
[](https://github.com/St4rG00se) \
 \
[](https://opensource.org/licenses/Apache-2.0)
## Table of contents
- [Quickstart](#quickstart)
- [Project template](#project-template)
- [Project links](#project-links)
- [Dev4py-utils modules](#dev4py-utils-modules)
- [dev4py.utils.AsyncJOptional](#dev4pyutilsasyncjoptional)
- [dev4py.utils.awaitables](#dev4pyutilsawaitables)
- [dev4py.utils.collectors](#dev4pyutilscollectors)
- [dev4py.utils.dicts](#dev4pyutilsdicts)
- [dev4py.utils.iterables](#dev4pyutilsiterables)
- [dev4py.utils.JOptional](#dev4pyutilsjoptional)
- [dev4py.utils.lists](#dev4pyutilslists)
- [dev4py.utils.objects](#dev4pyutilsobjects)
- [dev4py.utils.pipeline](#dev4pyutilspipeline)
- [dev4py.utils.retry](#dev4pyutilsretry)
- [dev4py.utils.Stream](#dev4pyutilsstream)
- [dev4py.utils.tuples](#dev4pyutilstuples)
- [dev4py.utils.types](#dev4pyutilstypes)
## Quickstart
```shell
pip install dev4py-utils
```
## Project template
This project is based on [pymsdl_template](https://github.com/dev4py/pymsdl_template)
## Project links
- [Documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils.html)
- [PyPi project](https://pypi.org/project/dev4py-utils/)
## Dev4py-utils modules
### dev4py.utils.AsyncJOptional
[AsyncJOptional documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/async_joptional.html)
> ***Note:** [AsyncJOptional](src/main/python/dev4py/utils/async_joptional.py) class is designed in order to simplify
> JOptional with async mapper.*
> ***Note:** AsyncJOptional support T or Awaitable[T] values. That's why some checks are done when terminal operation is
> called with `await`.*
Examples:
```python
import asyncio
from dev4py.utils import AsyncJOptional
def sync_mapper(i: int) -> int:
return i * 2
async def async_mapper(i: int) -> str:
return f"The value is {i}"
async def async_sample() -> None:
value: int = 1
await AsyncJOptional.of_noneable(value) \
.map(sync_mapper) \
.map(async_mapper) \
.if_present(print) # The value is 2
asyncio.run(async_sample())
```
### dev4py.utils.awaitables
[Awaitables documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/awaitables.html)
> ***Note:** [awaitables](src/main/python/dev4py/utils/awaitables.py) module provides a set of utility functions to
> simplify Awaitable operations.*
Examples:
```python
import asyncio
from dev4py.utils import awaitables, JOptional
# is_awaitable sample
awaitables.is_awaitable(asyncio.sleep(2)) # True
awaitables.is_awaitable(print('Hello')) # False
# to_sync_or_async_param_function sample
def mapper(s: str) -> str:
return s + '_suffix'
async def async_mapper(s: str) -> str:
await asyncio.sleep(1)
return s + '_async_suffix'
async def async_test():
# Note: mapper parameter is str and async_mapper returns an Awaitable[str] so we have to manage it
# Note: !WARNING! Since 3.0.0 see AsyncJOptional / JOptional to_async_joptional method
result: str = await JOptional.of("A value") \
.map(async_mapper) \
.map(awaitables.to_sync_or_async_param_function(mapper)) \
.get()
print(result) # A value_async_suffix_suffix
asyncio.run(async_test())
````
### dev4py.utils.collectors
[Collectors documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/collectors.html)
> ***Note:** The [collectors](src/main/python/dev4py/utils/collectors.py) class is inspired by
> [java.util.stream.Collectors](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/stream/Collectors.html)*
Examples:
```python
from dev4py.utils import Stream, collectors
Stream.of('a', 'b', 'c').collect(collectors.to_list()) # ['a', 'b', 'c']
```
### dev4py.utils.dicts
[Dicts documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/dicts.html)
> ***Note:** [dicts](src/main/python/dev4py/utils/dicts.py) module provides a set of utility functions to
> simplify dict operations.*
Examples:
```python
from dev4py.utils import dicts
from dev4py.utils.types import Supplier
# is_dict sample
dicts.is_dict("A str") # False
dicts.is_dict({'key': 'A dict value'}) # True
# get_value sample
int_supplier: Supplier[int] = lambda: 3
dictionary: dict[str, int] = {'key_1': 1, 'key_2': 2}
dicts.get_value(dictionary, 'key_1') # 1
dicts.get_value(dictionary, 'key_3') # None
dicts.get_value(dictionary, 'key_3', int_supplier) # 3
# get_value_from_path sample
str_supplier: Supplier[str] = lambda: "a3"
deep_dictionary: dict[str, dict[int, str]] = { \
'a': {1: 'a1', 2: 'a2'}, \
'b': {1: 'b1', 2: 'b2'} \
}
dicts.get_value_from_path(deep_dictionary, ["a", 1]) # 'a1'
dicts.get_value_from_path(deep_dictionary, ["c", 1]) # None
dicts.get_value_from_path(deep_dictionary, ["a", 3]) # None
dicts.get_value_from_path(deep_dictionary, ["a", 3], str_supplier) # 'a3'
````
### dev4py.utils.iterables
[Iterables documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/iterables.html)
> ***Note:** The [iterables](src/main/python/dev4py/utils/iterables.py) module provides a set of utility functions to simplify
> iterables operations.*
Example:
```python
from typing import Iterator
from dev4py.utils import iterables
values: range = range(0, 10)
chunks: Iterator[list[int]] = iterables.get_chunks(values, 3)
[chunk for chunk in chunks] # [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
```
### dev4py.utils.JOptional
[JOptional documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/joptional.html)
> ***Note:** [JOptional](src/main/python/dev4py/utils/joptional.py) class is inspired by
> [java.util.Optional](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/Optional.html)
> class with some adds (like `peek` method).*
Examples:
```python
from dev4py.utils import JOptional
value: int = 1
JOptional.of_noneable(value) \
.map(lambda v: f"The value is {v}") \
.if_present(print) # The value is 1
```
### dev4py.utils.lists
[Lists documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/lists.html)
> ***Note:** [lists](src/main/python/dev4py/utils/lists.py) module provides a set of utility functions to simplify lists
> operations.*
Examples:
```python
from dev4py.utils import lists
# empty sample
lst: list[int] = lists.empty_list() # []
# append sample
lst: list[int] = [1, 2, 3, 4]
app_lst: list[int] = lists.append(lst, 5) # [1, 2, 3, 4, 5]
# - Note: lst == app_lst
# extend sample
lst: list[int] = [1, 2, 3, 4]
lst2: list[int] = [5, 6, 7, 8]
ext_lst: list[int] = lists.extend(lst, lst2) # [1, 2, 3, 4, 5, 6, 7, 8]
# - Note: lst == ext_lst
```
### dev4py.utils.objects
[Objects documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/objects.html)
> ***Note:** The [objects](src/main/python/dev4py/utils/objects.py) module is inspired by
> [java.util.Objects](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/Objects.html)
> class.*
Examples:
```python
from dev4py.utils import objects
# non_none sample
value = None
objects.non_none(value) # False
# require_non_none sample
value = "A value"
objects.require_non_none(value) # 'A value'
# to_string sample
value = None
default_value: str = "A default value"
objects.to_string(value, default_value) # 'A default value'
```
### dev4py.utils.pipeline
[Pipeline documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/pipeline.html)
> ***Note:** The [pipeline](src/main/python/dev4py/utils/pipeline) package provides a set of Pipeline class describing
> different kind of pipelines.*
Examples:
```python
from dev4py.utils.pipeline import SimplePipeline, StepPipeline, StepResult
# SimplePipeline sample
pipeline: SimplePipeline[int, str] = SimplePipeline.of(lambda i: i * i) \
.add_handler(str) \
.add_handler(lambda s: f"Result: {s} | Type: {type(s)}")
pipeline.execute(10) # "Result: 100 | Type: <class 'str'>"
# StepPipeline sample
# Note: StepPipeline can be stopped at each step by setting `go_next` value to False
pipeline: StepPipeline[int, str] = StepPipeline.of(lambda i: StepResult(i * i)) \
.add_handler(lambda i: StepResult(value=str(i), go_next=i < 150)) \
.add_handler(lambda s: StepResult(f"Result: {s} | Type: {type(s)}"))
pipeline.execute(10) # StepResult(value="Result: 100 | Type: <class 'str'>", go_next=True)
# - Note: When the pipeline is fully completed, `go_next` is True
pipeline.execute(15) # StepResult(value='225', go_next=False)
# - Note: Even if the pipeline is not fully completed, the last StepResult is returned with `go_next=False`
```
### dev4py.utils.retry
[Retry documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/retry.html)
> ***Note:** The [retry](src/main/python/dev4py/utils/retry.py) module provides provides function to create retryable
> callable from simple sync or async callables using exponential backoff*
>
> *Usage idea: network requests (HTTP, AMQP, MQTT, etc.) with retry on error*
Examples:
```python
import asyncio
from time import time
from typing import Awaitable
from dev4py.utils.retry import RetryConfiguration, to_retryable, to_async_retryable, retryable, async_retryable
from dev4py.utils.types import BiFunction
# RetryConfiguration:
# Note: exponential backoff used formula is 'delay * (exponent^retry_number)'
#
# => Example: For the following RetryConfiguration, waiting times in case of error are:
# * first try: 0 sec (always 0 for the first try)
# * second try (/first retry): 1 sec ('0.5 * (2^1)')
# * third try (/second retry): 2 sec ('0.5 * (2^2)')
# * max_tries=3 => no fourth try (/third retry)
retry_config: RetryConfiguration = RetryConfiguration(
delay=0.5, # the exponential backoff delay in second (default: 0.1)
exponent=2, # the exponential backoff exponent to determine delay between each try (default: 2)
max_tries=3 # max try number (first try included) (default: 3, i.e.: first try and 2 retry)
)
# to_retryable sample:
# -> SUCCESSFUL CALL SAMPLE
def callable_sample(j: int, start_time: float) -> int:
print("callable_sample - call time: '%.2f'" % (time() - start_time))
return j ** 2
retryable_sample: BiFunction[int, float, int] = to_retryable(sync_callable=callable_sample, retry_config=retry_config)
# Note: Since 3.5.0 you can also use `retryable(sync_callable=callable_sample, retry_config=retry_config)`
result: int = retryable_sample(3, time()) # result = 9
# outputs:
# callable_sample - call time: '0.00'
# -> IN ERROR CALL SAMPLE
def in_error_callable_sample(j: int, start_time: float) -> int:
print("in_error_callable_sample - call time: '%.2f'" % (time() - start_time))
raise ValueError(j)
in_error_retryable_sample: BiFunction[int, float, int] = \
to_retryable(sync_callable=in_error_callable_sample, retry_config=retry_config)
# Note: Since 3.5.0 you can also use `retryable(sync_callable=in_error_callable_sample, retry_config=retry_config)`
# Note: By default the last raised exception is raised if max_tries is reach. You can change this behavior by setting
# the `on_failure` parameter
result: int = in_error_retryable_sample(3, time())
# outputs:
# in_error_callable_sample - call time: '0.00'
# in_error_callable_sample - call time: '1.00'
# in_error_callable_sample - call time: '3.00'
# ValueError: 3
#
# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting
# the `on_failure` parameter
# -> DECORATOR SAMPLE
@retryable(retry_config=retry_config)
def decorated_in_error_callable_sample(j: int, start_time: float) -> int:
print("decorated_in_error_callable_sample - call time: '%.2f'" % (time() - start_time))
raise ValueError(j)
result: int = decorated_in_error_callable_sample(3, time())
# outputs:
# decorated_in_error_callable_sample - call time: '0.00'
# decorated_in_error_callable_sample - call time: '1.00'
# decorated_in_error_callable_sample - call time: '3.00'
# ValueError: 3
#
# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting
# the `on_failure` parameter
# to_async_retryable sample:
# -> IN ERROR CALL ASYNC SAMPLE
async def in_error_async_callable_sample(j: int, start_time: float) -> int:
print("in_error_async_callable_sample - call time: '%.2f'" % (time() - start_time))
raise ValueError(j)
async def async_retryable_sample() -> None:
in_error_async_retryable_sample: BiFunction[int, float, Awaitable[int]] = \
to_async_retryable(async_callable=in_error_async_callable_sample, retry_config=retry_config)
# Note: Since 3.5.0 you can also use
# `async_retryable(async_callable=in_error_async_callable_sample, retry_config=retry_config)`
result: int = await in_error_async_retryable_sample(2, time())
asyncio.run(async_retryable_sample())
# outputs:
# in_error_async_callable_sample - call time: '0.00'
# in_error_async_callable_sample - call time: '1.00'
# in_error_async_callable_sample - call time: '3.00'
# ValueError: 2
#
# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting
# the `on_failure` parameter
# -> DECORATOR ASYNC SAMPLE
@async_retryable(retry_config=retry_config)
async def decorated_in_error_async_callable_sample(j: int, start_time: float) -> int:
print("decorated_in_error_async_callable_sample - call time: '%.2f'" % (time() - start_time))
raise ValueError(j)
async def async_decorated_retryable_sample() -> None:
result: int = await decorated_in_error_async_callable_sample(2, time())
asyncio.run(async_decorated_retryable_sample())
# outputs:
# in_error_async_callable_sample - call time: '0.00'
# in_error_async_callable_sample - call time: '1.00'
# in_error_async_callable_sample - call time: '3.00'
# ValueError: 2
#
# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting
# the `on_failure` parameter
```
### dev4py.utils.Stream
[Stream documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/stream.html)
> ***Note:** [Stream](src/main/python/dev4py/utils/stream.py) class is inspired by
> [java.util.stream.Stream](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/stream/Stream.html).*
Examples:
```python
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from time import sleep
from dev4py.utils import Stream, ParallelConfiguration
# Sequential sample
Stream.of(1, 2, 3, 4) \
.map(str) \
.peek(lambda s: sleep(0.5)) \
.map(lambda s: f"Mapped value: {s}") \
.to_list() # ['Mapped value: 1', 'Mapped value: 2', 'Mapped value: 3', 'Mapped value: 4']
# - Note: Execution time around 2 sec due to the sleep call
# Multithreading sample
with ThreadPoolExecutor(max_workers=2) as executor:
Stream.of(1, 2, 3, 4) \
.parallel(parallel_config=ParallelConfiguration(executor=executor, chunksize=2)) \
.map(str) \
.peek(lambda s: sleep(0.5)) \
.map(lambda s: f"Mapped value: {s}") \
.to_list() # ['Mapped value: 3', 'Mapped value: 4', 'Mapped value: 1', 'Mapped value: 2']
# - Note: Execution time around 1 sec due to the given ParallelConfiguration
# - Note: Since this stream is (by default) unordered, results order is random
# Multiprocessing sample
# - Note: Due to use of Multiprocessing:
# * lambdas cannot be used since they cannot be pickled
# * This sample should be put in a python file in order to work
def _sleep(s: str) -> None:
# eq lambda s: sleep(0.5)
sleep(0.5)
def _mapper(s: str) -> str:
# eq lambda s: f"Mapped value: {s}"
return f"Mapped value: {s}"
if __name__ == '__main__':
with ProcessPoolExecutor(max_workers=2) as executor:
Stream.of(1, 2, 3, 4) \
.parallel(parallel_config=ParallelConfiguration(executor=executor, chunksize=2)) \
.map(str) \
.peek(_sleep) \
.map(_mapper) \
.to_list()
# - Note: Execution time around 1 sec due to the given ParallelConfiguration
# (Reminder: Use Multiprocessing for CPU-bound tasks. In this case Multithreading is more appropriate)
# - Note: Since this stream is (by default) unordered, results order is random
```
### dev4py.utils.tuples
[Tuples documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/tuples.html)
> ***Note:** [tuples](src/main/python/dev4py/utils/tuples.py) module provides a set of utility functions to simplify
> tuples operations.*
Examples:
```python
from dev4py.utils import tuples
# empty sample
tpl: tuple[int, ...] = tuples.empty_tuple() # ()
# append sample
tpl: tuple[int, ...] = (1, 2, 3, 4)
app_tpl: tuple[int, ...] = tuples.append(tpl, 5) # (1, 2, 3, 4, 5)
# extend sample
tpl: tuple[int, ...] = (1, 2, 3, 4)
tpl2: tuple[int, ...] = (5, 6, 7, 8)
ext_tpl: tuple[int, ...] = tuples.extend(tpl, tpl2) # (1, 2, 3, 4, 5, 6, 7, 8)
```
### dev4py.utils.types
[Types documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/types.html)
> ***Note:** The [types](src/main/python/dev4py/utils/types.py) module is inspired by
> [java.util.function](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/function/package-summary.html)
> package.*
Examples:
```python
from dev4py.utils.types import Function, Predicate, Consumer
# Function sample
int_to_str: Function[int, str] = lambda i: str(i)
str_result: str = int_to_str(1) # '1'
# Predicate sample
str_predicate: Predicate[str] = lambda s: s == "A value"
pred_result: bool = str_predicate("Value to test") # False
# Consumer sample
def sample(consumer: Consumer[str], value: str) -> None:
consumer(value)
def my_consumer(arg: str) -> None:
print(arg)
sample(my_consumer, "My value") # My value
```
Raw data
{
"_id": null,
"home_page": "https://github.com/dev4py/dev4py-utils",
"name": "dev4py-utils",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.10,<4.0",
"maintainer_email": "",
"keywords": "python,utils,function,classes",
"author": "St4rG00se",
"author_email": "st4rg00se@protonmail.com",
"download_url": "https://files.pythonhosted.org/packages/fb/2f/5c4517e68def6932b517ba63bf6d06ccd455d1807e7df9d7cbbd08188e74/dev4py_utils-3.5.4.tar.gz",
"platform": null,
"description": "# Dev4py-utils\n\nA set of Python regularly used classes/functions\n\n[](https://github.com/dev4py/dev4py-utils/actions/workflows/ci.yml) \\\n[](https://github.com/dev4py/dev4py-utils/actions/workflows/on_release.yml) \\\n[](https://github.com/dev4py/dev4py-utils/actions/workflows/weekly_checks.yml) \\\n[](https://www.python.org/) \\\n[](https://github.com/St4rG00se) \\\n \\\n[](https://opensource.org/licenses/Apache-2.0)\n\n## Table of contents\n\n- [Quickstart](#quickstart)\n- [Project template](#project-template)\n- [Project links](#project-links)\n- [Dev4py-utils modules](#dev4py-utils-modules)\n - [dev4py.utils.AsyncJOptional](#dev4pyutilsasyncjoptional)\n - [dev4py.utils.awaitables](#dev4pyutilsawaitables)\n - [dev4py.utils.collectors](#dev4pyutilscollectors)\n - [dev4py.utils.dicts](#dev4pyutilsdicts)\n - [dev4py.utils.iterables](#dev4pyutilsiterables)\n - [dev4py.utils.JOptional](#dev4pyutilsjoptional)\n - [dev4py.utils.lists](#dev4pyutilslists)\n - [dev4py.utils.objects](#dev4pyutilsobjects)\n - [dev4py.utils.pipeline](#dev4pyutilspipeline)\n - [dev4py.utils.retry](#dev4pyutilsretry)\n - [dev4py.utils.Stream](#dev4pyutilsstream)\n - [dev4py.utils.tuples](#dev4pyutilstuples)\n - [dev4py.utils.types](#dev4pyutilstypes)\n\n## Quickstart\n\n```shell\npip install dev4py-utils\n```\n\n## Project template\n\nThis project is based on [pymsdl_template](https://github.com/dev4py/pymsdl_template)\n\n## Project links\n\n- [Documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils.html)\n- [PyPi project](https://pypi.org/project/dev4py-utils/)\n\n## Dev4py-utils modules\n\n### dev4py.utils.AsyncJOptional\n\n[AsyncJOptional documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/async_joptional.html)\n\n> ***Note:** [AsyncJOptional](src/main/python/dev4py/utils/async_joptional.py) class is designed in order to simplify\n> JOptional with async mapper.*\n\n> ***Note:** AsyncJOptional support T or Awaitable[T] values. That's why some checks are done when terminal operation is\n> called with `await`.*\n\nExamples:\n\n```python\nimport asyncio\nfrom dev4py.utils import AsyncJOptional\n\ndef sync_mapper(i: int) -> int:\n return i * 2\n\nasync def async_mapper(i: int) -> str:\n return f\"The value is {i}\"\n\nasync def async_sample() -> None:\n value: int = 1\n await AsyncJOptional.of_noneable(value) \\\n .map(sync_mapper) \\\n .map(async_mapper) \\\n .if_present(print) # The value is 2\n\nasyncio.run(async_sample())\n```\n\n### dev4py.utils.awaitables\n\n[Awaitables documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/awaitables.html)\n\n> ***Note:** [awaitables](src/main/python/dev4py/utils/awaitables.py) module provides a set of utility functions to\n> simplify Awaitable operations.*\n\nExamples:\n\n```python\nimport asyncio\nfrom dev4py.utils import awaitables, JOptional\n\n# is_awaitable sample\nawaitables.is_awaitable(asyncio.sleep(2)) # True\nawaitables.is_awaitable(print('Hello')) # False\n\n\n# to_sync_or_async_param_function sample\ndef mapper(s: str) -> str:\n return s + '_suffix'\n\nasync def async_mapper(s: str) -> str:\n await asyncio.sleep(1)\n return s + '_async_suffix'\n\nasync def async_test():\n # Note: mapper parameter is str and async_mapper returns an Awaitable[str] so we have to manage it\n # Note: !WARNING! Since 3.0.0 see AsyncJOptional / JOptional to_async_joptional method\n result: str = await JOptional.of(\"A value\") \\\n .map(async_mapper) \\\n .map(awaitables.to_sync_or_async_param_function(mapper)) \\\n .get()\n print(result) # A value_async_suffix_suffix\n\nasyncio.run(async_test())\n````\n\n### dev4py.utils.collectors\n\n[Collectors documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/collectors.html)\n\n> ***Note:** The [collectors](src/main/python/dev4py/utils/collectors.py) class is inspired by\n> [java.util.stream.Collectors](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/stream/Collectors.html)*\n\nExamples:\n\n```python\nfrom dev4py.utils import Stream, collectors\n\nStream.of('a', 'b', 'c').collect(collectors.to_list()) # ['a', 'b', 'c']\n```\n\n### dev4py.utils.dicts\n\n[Dicts documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/dicts.html)\n\n> ***Note:** [dicts](src/main/python/dev4py/utils/dicts.py) module provides a set of utility functions to\n> simplify dict operations.*\n\nExamples:\n\n```python\nfrom dev4py.utils import dicts\nfrom dev4py.utils.types import Supplier\n\n# is_dict sample\ndicts.is_dict(\"A str\") # False\ndicts.is_dict({'key': 'A dict value'}) # True\n\n\n# get_value sample\nint_supplier: Supplier[int] = lambda: 3\ndictionary: dict[str, int] = {'key_1': 1, 'key_2': 2}\n\ndicts.get_value(dictionary, 'key_1') # 1\ndicts.get_value(dictionary, 'key_3') # None\ndicts.get_value(dictionary, 'key_3', int_supplier) # 3\n\n\n# get_value_from_path sample\nstr_supplier: Supplier[str] = lambda: \"a3\"\ndeep_dictionary: dict[str, dict[int, str]] = { \\\n 'a': {1: 'a1', 2: 'a2'}, \\\n 'b': {1: 'b1', 2: 'b2'} \\\n}\n\ndicts.get_value_from_path(deep_dictionary, [\"a\", 1]) # 'a1'\ndicts.get_value_from_path(deep_dictionary, [\"c\", 1]) # None\ndicts.get_value_from_path(deep_dictionary, [\"a\", 3]) # None\ndicts.get_value_from_path(deep_dictionary, [\"a\", 3], str_supplier) # 'a3'\n````\n\n### dev4py.utils.iterables\n\n[Iterables documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/iterables.html)\n\n> ***Note:** The [iterables](src/main/python/dev4py/utils/iterables.py) module provides a set of utility functions to simplify\n> iterables operations.*\n\nExample:\n\n```python\nfrom typing import Iterator\n\nfrom dev4py.utils import iterables\n\nvalues: range = range(0, 10)\nchunks: Iterator[list[int]] = iterables.get_chunks(values, 3)\n[chunk for chunk in chunks] # [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]\n```\n\n### dev4py.utils.JOptional\n\n[JOptional documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/joptional.html)\n\n> ***Note:** [JOptional](src/main/python/dev4py/utils/joptional.py) class is inspired by\n> [java.util.Optional](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/Optional.html)\n> class with some adds (like `peek` method).*\n\nExamples:\n\n```python\nfrom dev4py.utils import JOptional\n\nvalue: int = 1\nJOptional.of_noneable(value) \\\n .map(lambda v: f\"The value is {v}\") \\\n .if_present(print) # The value is 1\n```\n\n### dev4py.utils.lists\n\n[Lists documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/lists.html)\n\n> ***Note:** [lists](src/main/python/dev4py/utils/lists.py) module provides a set of utility functions to simplify lists\n> operations.*\n\nExamples:\n\n```python\nfrom dev4py.utils import lists\n\n# empty sample\nlst: list[int] = lists.empty_list() # []\n\n# append sample\nlst: list[int] = [1, 2, 3, 4]\napp_lst: list[int] = lists.append(lst, 5) # [1, 2, 3, 4, 5]\n# - Note: lst == app_lst\n\n# extend sample\nlst: list[int] = [1, 2, 3, 4]\nlst2: list[int] = [5, 6, 7, 8]\next_lst: list[int] = lists.extend(lst, lst2) # [1, 2, 3, 4, 5, 6, 7, 8]\n# - Note: lst == ext_lst\n```\n\n### dev4py.utils.objects\n\n[Objects documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/objects.html)\n\n> ***Note:** The [objects](src/main/python/dev4py/utils/objects.py) module is inspired by\n> [java.util.Objects](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/Objects.html)\n> class.*\n\nExamples:\n\n```python\nfrom dev4py.utils import objects\n\n# non_none sample\nvalue = None\nobjects.non_none(value) # False\n\n# require_non_none sample\nvalue = \"A value\"\nobjects.require_non_none(value) # 'A value'\n\n# to_string sample\nvalue = None\ndefault_value: str = \"A default value\"\nobjects.to_string(value, default_value) # 'A default value'\n```\n\n### dev4py.utils.pipeline\n\n[Pipeline documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/pipeline.html)\n\n> ***Note:** The [pipeline](src/main/python/dev4py/utils/pipeline) package provides a set of Pipeline class describing\n> different kind of pipelines.*\n\nExamples:\n\n```python\nfrom dev4py.utils.pipeline import SimplePipeline, StepPipeline, StepResult\n\n# SimplePipeline sample\npipeline: SimplePipeline[int, str] = SimplePipeline.of(lambda i: i * i) \\\n .add_handler(str) \\\n .add_handler(lambda s: f\"Result: {s} | Type: {type(s)}\")\n\npipeline.execute(10) # \"Result: 100 | Type: <class 'str'>\"\n\n\n# StepPipeline sample\n# Note: StepPipeline can be stopped at each step by setting `go_next` value to False\npipeline: StepPipeline[int, str] = StepPipeline.of(lambda i: StepResult(i * i)) \\\n .add_handler(lambda i: StepResult(value=str(i), go_next=i < 150)) \\\n .add_handler(lambda s: StepResult(f\"Result: {s} | Type: {type(s)}\"))\n\npipeline.execute(10) # StepResult(value=\"Result: 100 | Type: <class 'str'>\", go_next=True)\n# - Note: When the pipeline is fully completed, `go_next` is True\npipeline.execute(15) # StepResult(value='225', go_next=False)\n# - Note: Even if the pipeline is not fully completed, the last StepResult is returned with `go_next=False`\n```\n\n### dev4py.utils.retry\n\n[Retry documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/retry.html)\n\n> ***Note:** The [retry](src/main/python/dev4py/utils/retry.py) module provides provides function to create retryable\n> callable from simple sync or async callables using exponential backoff*\n>\n> *Usage idea: network requests (HTTP, AMQP, MQTT, etc.) with retry on error*\n\nExamples:\n\n```python\nimport asyncio\nfrom time import time\nfrom typing import Awaitable\n\nfrom dev4py.utils.retry import RetryConfiguration, to_retryable, to_async_retryable, retryable, async_retryable\nfrom dev4py.utils.types import BiFunction\n\n# RetryConfiguration:\n# Note: exponential backoff used formula is 'delay * (exponent^retry_number)'\n#\n# => Example: For the following RetryConfiguration, waiting times in case of error are:\n# * first try: 0 sec (always 0 for the first try)\n# * second try (/first retry): 1 sec ('0.5 * (2^1)')\n# * third try (/second retry): 2 sec ('0.5 * (2^2)')\n# * max_tries=3 => no fourth try (/third retry)\nretry_config: RetryConfiguration = RetryConfiguration(\n delay=0.5, # the exponential backoff delay in second (default: 0.1)\n exponent=2, # the exponential backoff exponent to determine delay between each try (default: 2)\n max_tries=3 # max try number (first try included) (default: 3, i.e.: first try and 2 retry)\n)\n\n\n# to_retryable sample:\n# -> SUCCESSFUL CALL SAMPLE\ndef callable_sample(j: int, start_time: float) -> int:\n print(\"callable_sample - call time: '%.2f'\" % (time() - start_time))\n return j ** 2\n\nretryable_sample: BiFunction[int, float, int] = to_retryable(sync_callable=callable_sample, retry_config=retry_config)\n# Note: Since 3.5.0 you can also use `retryable(sync_callable=callable_sample, retry_config=retry_config)`\n\nresult: int = retryable_sample(3, time()) # result = 9\n# outputs:\n# callable_sample - call time: '0.00'\n\n\n# -> IN ERROR CALL SAMPLE\ndef in_error_callable_sample(j: int, start_time: float) -> int:\n print(\"in_error_callable_sample - call time: '%.2f'\" % (time() - start_time))\n raise ValueError(j)\n\nin_error_retryable_sample: BiFunction[int, float, int] = \\\n to_retryable(sync_callable=in_error_callable_sample, retry_config=retry_config)\n# Note: Since 3.5.0 you can also use `retryable(sync_callable=in_error_callable_sample, retry_config=retry_config)`\n# Note: By default the last raised exception is raised if max_tries is reach. You can change this behavior by setting\n# the `on_failure` parameter\nresult: int = in_error_retryable_sample(3, time())\n# outputs:\n# in_error_callable_sample - call time: '0.00'\n# in_error_callable_sample - call time: '1.00'\n# in_error_callable_sample - call time: '3.00'\n# ValueError: 3\n#\n# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting\n# the `on_failure` parameter\n\n# -> DECORATOR SAMPLE\n@retryable(retry_config=retry_config)\ndef decorated_in_error_callable_sample(j: int, start_time: float) -> int:\n print(\"decorated_in_error_callable_sample - call time: '%.2f'\" % (time() - start_time))\n raise ValueError(j)\n\nresult: int = decorated_in_error_callable_sample(3, time())\n# outputs:\n# decorated_in_error_callable_sample - call time: '0.00'\n# decorated_in_error_callable_sample - call time: '1.00'\n# decorated_in_error_callable_sample - call time: '3.00'\n# ValueError: 3\n#\n# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting\n# the `on_failure` parameter\n\n\n# to_async_retryable sample:\n# -> IN ERROR CALL ASYNC SAMPLE\nasync def in_error_async_callable_sample(j: int, start_time: float) -> int:\n print(\"in_error_async_callable_sample - call time: '%.2f'\" % (time() - start_time))\n raise ValueError(j)\n\nasync def async_retryable_sample() -> None:\n in_error_async_retryable_sample: BiFunction[int, float, Awaitable[int]] = \\\n to_async_retryable(async_callable=in_error_async_callable_sample, retry_config=retry_config)\n # Note: Since 3.5.0 you can also use \n # `async_retryable(async_callable=in_error_async_callable_sample, retry_config=retry_config)`\n result: int = await in_error_async_retryable_sample(2, time())\n\nasyncio.run(async_retryable_sample())\n# outputs:\n# in_error_async_callable_sample - call time: '0.00'\n# in_error_async_callable_sample - call time: '1.00'\n# in_error_async_callable_sample - call time: '3.00'\n# ValueError: 2\n#\n# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting\n# the `on_failure` parameter\n\n# -> DECORATOR ASYNC SAMPLE\n@async_retryable(retry_config=retry_config)\nasync def decorated_in_error_async_callable_sample(j: int, start_time: float) -> int:\n print(\"decorated_in_error_async_callable_sample - call time: '%.2f'\" % (time() - start_time))\n raise ValueError(j)\n\nasync def async_decorated_retryable_sample() -> None:\n result: int = await decorated_in_error_async_callable_sample(2, time())\n\nasyncio.run(async_decorated_retryable_sample())\n# outputs:\n# in_error_async_callable_sample - call time: '0.00'\n# in_error_async_callable_sample - call time: '1.00'\n# in_error_async_callable_sample - call time: '3.00'\n# ValueError: 2\n#\n# Note: By default the last raised exception is raised if max_tries is reached. You can change this behavior by setting\n# the `on_failure` parameter\n```\n\n### dev4py.utils.Stream\n\n[Stream documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/stream.html)\n\n> ***Note:** [Stream](src/main/python/dev4py/utils/stream.py) class is inspired by\n> [java.util.stream.Stream](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/stream/Stream.html).*\n\nExamples:\n\n```python\nfrom concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor\nfrom time import sleep\n\nfrom dev4py.utils import Stream, ParallelConfiguration\n\n# Sequential sample\nStream.of(1, 2, 3, 4) \\\n .map(str) \\\n .peek(lambda s: sleep(0.5)) \\\n .map(lambda s: f\"Mapped value: {s}\") \\\n .to_list() # ['Mapped value: 1', 'Mapped value: 2', 'Mapped value: 3', 'Mapped value: 4']\n# - Note: Execution time around 2 sec due to the sleep call\n\n\n# Multithreading sample\nwith ThreadPoolExecutor(max_workers=2) as executor:\n Stream.of(1, 2, 3, 4) \\\n .parallel(parallel_config=ParallelConfiguration(executor=executor, chunksize=2)) \\\n .map(str) \\\n .peek(lambda s: sleep(0.5)) \\\n .map(lambda s: f\"Mapped value: {s}\") \\\n .to_list() # ['Mapped value: 3', 'Mapped value: 4', 'Mapped value: 1', 'Mapped value: 2']\n# - Note: Execution time around 1 sec due to the given ParallelConfiguration\n# - Note: Since this stream is (by default) unordered, results order is random\n\n\n# Multiprocessing sample\n# - Note: Due to use of Multiprocessing:\n# * lambdas cannot be used since they cannot be pickled\n# * This sample should be put in a python file in order to work\ndef _sleep(s: str) -> None:\n # eq lambda s: sleep(0.5)\n sleep(0.5)\n\ndef _mapper(s: str) -> str:\n # eq lambda s: f\"Mapped value: {s}\"\n return f\"Mapped value: {s}\"\n\nif __name__ == '__main__':\n with ProcessPoolExecutor(max_workers=2) as executor:\n Stream.of(1, 2, 3, 4) \\\n .parallel(parallel_config=ParallelConfiguration(executor=executor, chunksize=2)) \\\n .map(str) \\\n .peek(_sleep) \\\n .map(_mapper) \\\n .to_list()\n\n# - Note: Execution time around 1 sec due to the given ParallelConfiguration\n# (Reminder: Use Multiprocessing for CPU-bound tasks. In this case Multithreading is more appropriate)\n# - Note: Since this stream is (by default) unordered, results order is random\n```\n\n### dev4py.utils.tuples\n\n[Tuples documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/tuples.html)\n\n> ***Note:** [tuples](src/main/python/dev4py/utils/tuples.py) module provides a set of utility functions to simplify\n> tuples operations.*\n\nExamples:\n\n```python\nfrom dev4py.utils import tuples\n\n# empty sample\ntpl: tuple[int, ...] = tuples.empty_tuple() # ()\n\n# append sample\ntpl: tuple[int, ...] = (1, 2, 3, 4)\napp_tpl: tuple[int, ...] = tuples.append(tpl, 5) # (1, 2, 3, 4, 5)\n\n# extend sample\ntpl: tuple[int, ...] = (1, 2, 3, 4)\ntpl2: tuple[int, ...] = (5, 6, 7, 8)\next_tpl: tuple[int, ...] = tuples.extend(tpl, tpl2) # (1, 2, 3, 4, 5, 6, 7, 8)\n```\n\n### dev4py.utils.types\n\n[Types documentation](https://htmlpreview.github.io/?https://github.com/dev4py/dev4py-utils/blob/main/docs/dev4py/utils/types.html)\n\n> ***Note:** The [types](src/main/python/dev4py/utils/types.py) module is inspired by\n> [java.util.function](https://docs.oracle.com/en/java/javase/17/docs/api//java.base/java/util/function/package-summary.html)\n> package.*\n\nExamples:\n\n```python\nfrom dev4py.utils.types import Function, Predicate, Consumer\n\n# Function sample\nint_to_str: Function[int, str] = lambda i: str(i)\nstr_result: str = int_to_str(1) # '1'\n\n# Predicate sample\nstr_predicate: Predicate[str] = lambda s: s == \"A value\"\npred_result: bool = str_predicate(\"Value to test\") # False\n\n# Consumer sample\ndef sample(consumer: Consumer[str], value: str) -> None:\n consumer(value)\n\ndef my_consumer(arg: str) -> None:\n print(arg)\n\nsample(my_consumer, \"My value\") # My value\n```\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "A set of Python regularly used classes/functions",
"version": "3.5.4",
"split_keywords": [
"python",
"utils",
"function",
"classes"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "1ba7169bf6dc561411cd4a78b66d8813e1d1568fd19b83251c16aa938d860e91",
"md5": "bc0311247785c14ab4f1f5d062227680",
"sha256": "5cd785fb9819bb6df1bd3acfd32eea3d143a3211915fb19997081e30d8c10855"
},
"downloads": -1,
"filename": "dev4py_utils-3.5.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "bc0311247785c14ab4f1f5d062227680",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10,<4.0",
"size": 41886,
"upload_time": "2023-01-24T11:37:28",
"upload_time_iso_8601": "2023-01-24T11:37:28.151826Z",
"url": "https://files.pythonhosted.org/packages/1b/a7/169bf6dc561411cd4a78b66d8813e1d1568fd19b83251c16aa938d860e91/dev4py_utils-3.5.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "fb2f5c4517e68def6932b517ba63bf6d06ccd455d1807e7df9d7cbbd08188e74",
"md5": "4d0e86d1cc67baa40106f0619205c39d",
"sha256": "ae41aad1b6a4cd0a58eca19b96cdb6299bdc5ab50160edfe413d788a2bf99cd3"
},
"downloads": -1,
"filename": "dev4py_utils-3.5.4.tar.gz",
"has_sig": false,
"md5_digest": "4d0e86d1cc67baa40106f0619205c39d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10,<4.0",
"size": 68163,
"upload_time": "2023-01-24T11:37:29",
"upload_time_iso_8601": "2023-01-24T11:37:29.652983Z",
"url": "https://files.pythonhosted.org/packages/fb/2f/5c4517e68def6932b517ba63bf6d06ccd455d1807e7df9d7cbbd08188e74/dev4py_utils-3.5.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-01-24 11:37:29",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "dev4py",
"github_project": "dev4py-utils",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "dev4py-utils"
}
JSON
