# Typesafe parmap
[](https://pypi.org/project/typesafe-parmap)
[](https://pypi.org/project/typesafe-parmap)
[](https://github.com/thejaminator/typesafe_parmap/actions/workflows/dev.yml)
```
pip install typesafe-parmap
```
Run functions in parallel safely with your type checkers
* GitHub: <https://github.com/thejaminator/typesafe_parmap>
## Features
Easy run different functions in parallel
```python
from typesafe_parmap import par_map_2
import time
from concurrent.futures import ThreadPoolExecutor
tp = ThreadPoolExecutor(5)
def long_running_int(param: int) -> int:
time.sleep(5) # long IO task here
return 123
def long_running_str(param: str) -> str:
time.sleep(5) # long IO task here
return "hello world"
int_result, str_result = par_map_2(
lambda: long_running_int(5),
lambda: long_running_str("test"),
executor=tp)
assert int_result == 123, str_result == "hello world" # should finish in around 5 seconds
```
Function return types are inferred correctly by mypy / pycharm
```python
reveal_type(int_result) # mypy infers int
reveal_type(str_result) # mypy infers str
```
Accidentally unpacked too many / little values? Type inference checks that for you!
```python
one, two, three, four = par_map_3(
lambda: long_running_int(5), lambda: long_running_str("test"), lambda: "something", executor=tp
) # Error: Need more than 3 values to unapck, (4 expected)
```
Got more than a few functions to run? We got you covered...
```python
from typesafe_parmap import par_map_4 # ... all the way to par_map_22!
```
Want to change the number of functions to run in parallel? Hate having to import a different one each time?
Use par_map_n!
```python
from typesafe_parmap import par_map_2, par_map_3, par_map_n
a = par_map_2(lambda: long_running_int(5), lambda: long_running_str("test"), executor=executor)
b = par_map_n(lambda: long_running_int(5), lambda: long_running_str("test"), executor=executor)
assert a == b
c = par_map_3(
lambda: long_running_int(5),
lambda: long_running_str("test"),
lambda: long_running_str("test"),
executor=executor,
)
d = par_map_n(
lambda: long_running_int(5),
lambda: long_running_str("test"),
lambda: long_running_str("test"),
executor=executor,
)
assert c == d
```
## Timeouts
Suppose you want to run a bunch of functions that might take a long time, but you don't want to wait forever.
Use par_map_timeout_n!
```python
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
from typesafe_parmap import par_map_timeout_n
# Since there are 3 threads, we should be able to run 3 functions at once
executor = ThreadPoolExecutor(3)
int_result, str_result_1, str_result_2 = par_map_timeout_n(
lambda: long_running_int(5),
lambda: short_running_str("test 1"),
lambda: short_running_str("test 2"),
executor=executor,
timeout=timedelta(seconds=5),
)
assert int_result is None # This function timed out
assert str_result_1 == "test 1" # This still finished in time
assert str_result_2 == "test 2" # This still finished in time
```
Note that as a result of the timeout, the return types of the int_result and str_result_1 are now Optional[str] and Optional[int] respectively.
### Logging timeouts
par_map_timeout_n accepts a logger parameter.
We also provide a class `NamedThunk`, which allows you to name your thunks so that the name is not just `<lambda>` in the logs.
```python
from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta
from typesafe_parmap import par_map_timeout_n, NamedThunk
executor = ThreadPoolExecutor(2)
par_map_timeout_n(
NamedThunk(lambda: long_running_int(5), name="Long Running Int"),
lambda: short_running_str("test 2"),
executor=executor,
timeout=timedelta(seconds=3),
logger=print,
)
# Prints:
# par_map func1: Long Running Int timed out after 3 seconds
```
Raw data
{
"_id": null,
"home_page": "https://github.com/thejaminator/typesafe_parmap",
"name": "typesafe-parmap",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "",
"author": "James Chua",
"author_email": "chuajamessh@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/56/b0/b6887d178406c11341253437444ddaf0e6e4c8c48c9e856103e6b5133831/typesafe_parmap-1.0.7.tar.gz",
"platform": null,
"description": "# Typesafe parmap\n\n\n[](https://pypi.org/project/typesafe-parmap)\n[](https://pypi.org/project/typesafe-parmap)\n[](https://github.com/thejaminator/typesafe_parmap/actions/workflows/dev.yml)\n\n```\npip install typesafe-parmap\n```\n\nRun functions in parallel safely with your type checkers\n\n\n* GitHub: <https://github.com/thejaminator/typesafe_parmap>\n\n\n## Features\n\nEasy run different functions in parallel\n```python\nfrom typesafe_parmap import par_map_2\nimport time\nfrom concurrent.futures import ThreadPoolExecutor\n\ntp = ThreadPoolExecutor(5)\n\ndef long_running_int(param: int) -> int:\n time.sleep(5) # long IO task here\n return 123\n\ndef long_running_str(param: str) -> str:\n time.sleep(5) # long IO task here\n return \"hello world\"\n\nint_result, str_result = par_map_2(\n lambda: long_running_int(5),\n lambda: long_running_str(\"test\"),\n executor=tp)\nassert int_result == 123, str_result == \"hello world\" # should finish in around 5 seconds\n```\n\nFunction return types are inferred correctly by mypy / pycharm\n\n```python\nreveal_type(int_result) # mypy infers int\nreveal_type(str_result) # mypy infers str\n```\n\nAccidentally unpacked too many / little values? Type inference checks that for you!\n```python\none, two, three, four = par_map_3(\n lambda: long_running_int(5), lambda: long_running_str(\"test\"), lambda: \"something\", executor=tp\n ) # Error: Need more than 3 values to unapck, (4 expected)\n```\n\nGot more than a few functions to run? We got you covered...\n```python\nfrom typesafe_parmap import par_map_4 # ... all the way to par_map_22!\n```\n\nWant to change the number of functions to run in parallel? Hate having to import a different one each time?\nUse par_map_n!\n```python\nfrom typesafe_parmap import par_map_2, par_map_3, par_map_n\na = par_map_2(lambda: long_running_int(5), lambda: long_running_str(\"test\"), executor=executor)\nb = par_map_n(lambda: long_running_int(5), lambda: long_running_str(\"test\"), executor=executor)\n\nassert a == b\n\nc = par_map_3(\n lambda: long_running_int(5),\n lambda: long_running_str(\"test\"),\n lambda: long_running_str(\"test\"),\n executor=executor,\n)\nd = par_map_n(\n lambda: long_running_int(5),\n lambda: long_running_str(\"test\"),\n lambda: long_running_str(\"test\"),\n executor=executor,\n)\n\nassert c == d\n```\n\n## Timeouts\nSuppose you want to run a bunch of functions that might take a long time, but you don't want to wait forever.\nUse par_map_timeout_n!\n```python\nfrom concurrent.futures import ThreadPoolExecutor\nfrom datetime import timedelta\nfrom typesafe_parmap import par_map_timeout_n\n# Since there are 3 threads, we should be able to run 3 functions at once\nexecutor = ThreadPoolExecutor(3)\nint_result, str_result_1, str_result_2 = par_map_timeout_n(\n lambda: long_running_int(5),\n lambda: short_running_str(\"test 1\"),\n lambda: short_running_str(\"test 2\"),\n executor=executor,\n timeout=timedelta(seconds=5),\n)\nassert int_result is None # This function timed out\nassert str_result_1 == \"test 1\" # This still finished in time\nassert str_result_2 == \"test 2\" # This still finished in time\n```\nNote that as a result of the timeout, the return types of the int_result and str_result_1 are now Optional[str] and Optional[int] respectively.\n\n\n### Logging timeouts\npar_map_timeout_n accepts a logger parameter.\nWe also provide a class `NamedThunk`, which allows you to name your thunks so that the name is not just `<lambda>` in the logs.\n```python\nfrom concurrent.futures import ThreadPoolExecutor\nfrom datetime import timedelta\nfrom typesafe_parmap import par_map_timeout_n, NamedThunk\nexecutor = ThreadPoolExecutor(2)\npar_map_timeout_n(\n NamedThunk(lambda: long_running_int(5), name=\"Long Running Int\"),\n lambda: short_running_str(\"test 2\"),\n executor=executor,\n timeout=timedelta(seconds=3),\n logger=print,\n)\n# Prints:\n# par_map func1: Long Running Int timed out after 3 seconds\n```\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Run functions in parallel safely with typesafe parmap!.",
"version": "1.0.7",
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"md5": "a1b9a380d4809c32832e705b14be2316",
"sha256": "4ea6bc53e2d37ad1d8e9880cbd4e53f49f1afc610fd398291840af73d831b9d0"
},
"downloads": -1,
"filename": "typesafe_parmap-1.0.7-py3-none-any.whl",
"has_sig": false,
"md5_digest": "a1b9a380d4809c32832e705b14be2316",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 11226,
"upload_time": "2022-12-20T07:35:30",
"upload_time_iso_8601": "2022-12-20T07:35:30.656419Z",
"url": "https://files.pythonhosted.org/packages/be/bf/e3e8ee75c754078ce1406a761e32b74d2ac63a16538dd323799189c1525b/typesafe_parmap-1.0.7-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"md5": "fd2f309d5d72e2020322c74d74c31c8c",
"sha256": "ed0b887b4709676debbf35a7413c37d711187e21d094ccc437aff72aeda2b0ec"
},
"downloads": -1,
"filename": "typesafe_parmap-1.0.7.tar.gz",
"has_sig": false,
"md5_digest": "fd2f309d5d72e2020322c74d74c31c8c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 13522,
"upload_time": "2022-12-20T07:35:32",
"upload_time_iso_8601": "2022-12-20T07:35:32.412951Z",
"url": "https://files.pythonhosted.org/packages/56/b0/b6887d178406c11341253437444ddaf0e6e4c8c48c9e856103e6b5133831/typesafe_parmap-1.0.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2022-12-20 07:35:32",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "thejaminator",
"github_project": "typesafe_parmap",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "typesafe-parmap"
}
JSON
