[](https://app.codecov.io/github/mreiche/python-streams)
# tinystream / python-streams
This is a simple and lightweight Streams API inspired by Java Streams with support for type hinting.
This package is released as `tinystream` at pypi.
## Basic API
```python
from tinystream import Stream
stream = Stream([1, 2, 3, 4, 5]) # Stream.of_many(*), Stream.of_dict()
stream \
.map(lambda x: x + 1) \ # flatmap(), peek(), map_key(), map_kwargs(), map_keys()
.filter(lambda x: x > 2) \ # filter_key(), filter_type()
.sorted(reverse=True) \ # sort()
.reverse() \
.limit(2) \
.concat([4]) \
.sum() # reduce(), max(), min(), collect(), count(), find()
```
## Aggregators
Aggregators like `sum()`, `count()`, `max()` will `collect()` the data and end the stream. `collect()` also caches the data and can be called multiple times, since it returns only a `list`.
## Built-in Optional support
Some aggregators like `sum()`, `max()` are *optional*:
```python
assert Stream((1, 2, 3, 4, 5)).sum().present
```
Get next value as *optional*:
```python
assert Stream((1, 2, 3, 4, 5)).next().present
```
Create custom *optional*:
```python
from tinystream import Opt
assert Opt(None).absent
```
Map *optional*:
```python
assert Opt("String").map(str.lower).len == 6
```
Get default value:
```python
assert Opt(None).get(6) == 6
assert Opt(None).get(lambda: 6) == 6
assert Opt(None).if_absent(lambda: 3).present
```
Filter value:
```python
assert Opt(0).filter(lambda x: x > 0).absent
```
You can also access optional index elements of the stream, but this will `collect()` and end the stream.
```python
assert Stream([])[2].absent
```
## Type hinting
You can typehint datatypes like:
```python
from dataclasses import dataclass
@dataclass
class Node:
name: str
parent: "Node" = None
parent = Node(name="B")
child = Node(name="A", parent=parent)
```
for lambdas:
```python
stream = Stream([child])
assert stream.map(lambda x: x.parent).type(Node).next().get().name == "B"
```
This is not necessary when you pass a mapping function:
```python
def map_parent(n: Node):
return n.parent
assert stream.map(map_parent).next().get().name == "B"
```
## More features
### Typed dictionaries
Dictionaries are streamed as `tuple(key, value)`
```python
children = {"a": Node(name="Child")}
stream = Stream.of_dict(children)
for item in stream:
# item[0] is known as str
# item[1] is known as Node
```
This is the same like (but without known types):
```python
stream = Stream(children)
```
### Filter by existing key
```python
items_with_name = Stream([child]).filter_key("name")
```
### Filter by type
```python
nodes_only = Stream([child]).filter_type(Node)
```
### Map object name attribute
```python
names = Stream([child]).map_key("name")
```
### Deep mapping of name attributes
```python
list = [
{"node": Node(name="Node A")},
{"node": Node(name="Node B")},
{"node": Node(name="Node C")},
{"node": Node(name="Node D")},
]
names = Stream(list).map_keys("node", "name")
```
### Collected join
```python
all_names = Stream([child]).map_key("name").join(", ")
```
### Map kwargs
```python
list = [
{"name": "Node A"},
{"name": "Node B"},
]
# Short cut for map(lambda x: Node(**x))
nodes = Stream(list).map_kwargs(Node)
```
### Stream many
```python
many = Stream.of_many([1, 2, 3], (4, 5, 6))
many = many.concat([7, 8, 9])
```
### End of stream
```python
stream = Stream(["a", "b", "c"]).on_end(lambda: print("Finished"))
char = stream.next().get()
if char == "a":
stream.end()
```
## Examples
A given data structure like:
```python
data = {
"ranges": [
{"days": 3},
{"weeks": 1},
]
}
```
Without tinystream:
```python
if "ranges" in data:
range_data: timedelta
for range_data in map(lambda x: timedelta(**x), data["ranges"]):
pass
```
With tinystream:
```python
for range_data in Opt(data).map_key("ranges").stream().map_kwargs(timedelta):
pass
```
## Comparison with other libraries
There are a couple of other implementation to fulfill similar requirements.
- https://github.com/vxgmichel/aiostream
- https://github.com/python-streamz/streamz
- https://pypi.org/project/fluentpy
- https://github.com/ramsteak/streams
- https://github.com/alemazzo/Python-Java-Stream (*outdated*)
- https://github.com/JaviOverflow/python-streams (*outdated*)
- https://github.com/9seconds/streams/ (*outdated*)
- https://github.com/tolsac/streampy (*outdated*)
- Apache Spark
## Run the tests
```shell
PYTHONPATH="." pytest --cov=tinystream -n 4 tests/
```
### Release update
1. Update version in `setup.py`
2. Package library
```shell
python setup.py sdist
```
3. Publish library
```shell
twine upload dist/tinystream-[version].tar.gz
```
## References
- https://github.com/MichaelKim0407/tutorial-pip-package
- https://packaging.python.org/en/latest/guides/making-a-pypi-friendly-readme/
Raw data
{
"_id": null,
"home_page": "https://github.com/mreiche/python-streams",
"name": "tinystream",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": null,
"author": "Mike Reiche",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/a1/6e/94cc00b45af7f3ea28122fa444fd33b94823e2bb23e9f392fbc99d5374b6/tinystream-0.1.15.tar.gz",
"platform": null,
"description": "\n[](https://app.codecov.io/github/mreiche/python-streams)\n\n# tinystream / python-streams\n\nThis is a simple and lightweight Streams API inspired by Java Streams with support for type hinting.\n\nThis package is released as `tinystream` at pypi.\n\n## Basic API\n\n```python\nfrom tinystream import Stream\n\nstream = Stream([1, 2, 3, 4, 5]) # Stream.of_many(*), Stream.of_dict()\n\nstream \\\n .map(lambda x: x + 1) \\ # flatmap(), peek(), map_key(), map_kwargs(), map_keys()\n .filter(lambda x: x > 2) \\ # filter_key(), filter_type()\n .sorted(reverse=True) \\ # sort()\n .reverse() \\\n .limit(2) \\\n .concat([4]) \\\n .sum() # reduce(), max(), min(), collect(), count(), find()\n```\n\n## Aggregators\n\nAggregators like `sum()`, `count()`, `max()` will `collect()` the data and end the stream. `collect()` also caches the data and can be called multiple times, since it returns only a `list`.\n\n## Built-in Optional support\n\nSome aggregators like `sum()`, `max()` are *optional*:\n\n```python\nassert Stream((1, 2, 3, 4, 5)).sum().present\n```\n\nGet next value as *optional*:\n\n```python\nassert Stream((1, 2, 3, 4, 5)).next().present\n```\n\nCreate custom *optional*:\n\n```python\nfrom tinystream import Opt\n\nassert Opt(None).absent\n```\n\nMap *optional*:\n```python\nassert Opt(\"String\").map(str.lower).len == 6\n```\n\nGet default value:\n```python\nassert Opt(None).get(6) == 6\nassert Opt(None).get(lambda: 6) == 6\nassert Opt(None).if_absent(lambda: 3).present\n```\n\nFilter value:\n\n```python\nassert Opt(0).filter(lambda x: x > 0).absent\n```\n\nYou can also access optional index elements of the stream, but this will `collect()` and end the stream.\n```python\nassert Stream([])[2].absent\n```\n\n## Type hinting\n\nYou can typehint datatypes like:\n\n```python\nfrom dataclasses import dataclass\n\n@dataclass\nclass Node:\n name: str\n parent: \"Node\" = None\n\nparent = Node(name=\"B\")\nchild = Node(name=\"A\", parent=parent)\n```\n\nfor lambdas:\n\n```python\nstream = Stream([child])\nassert stream.map(lambda x: x.parent).type(Node).next().get().name == \"B\"\n```\n\nThis is not necessary when you pass a mapping function:\n```python\ndef map_parent(n: Node):\n return n.parent\n\nassert stream.map(map_parent).next().get().name == \"B\"\n```\n\n## More features\n\n### Typed dictionaries\n\nDictionaries are streamed as `tuple(key, value)`\n\n```python\nchildren = {\"a\": Node(name=\"Child\")} \nstream = Stream.of_dict(children)\nfor item in stream:\n # item[0] is known as str\n # item[1] is known as Node\n```\n\nThis is the same like (but without known types):\n```python\nstream = Stream(children)\n```\n\n### Filter by existing key\n```python\nitems_with_name = Stream([child]).filter_key(\"name\")\n```\n\n### Filter by type\n```python\nnodes_only = Stream([child]).filter_type(Node)\n```\n\n### Map object name attribute\n```python\nnames = Stream([child]).map_key(\"name\")\n```\n\n### Deep mapping of name attributes\n```python\nlist = [\n {\"node\": Node(name=\"Node A\")},\n {\"node\": Node(name=\"Node B\")},\n {\"node\": Node(name=\"Node C\")},\n {\"node\": Node(name=\"Node D\")},\n]\nnames = Stream(list).map_keys(\"node\", \"name\")\n```\n\n### Collected join\n\n```python\nall_names = Stream([child]).map_key(\"name\").join(\", \")\n```\n\n### Map kwargs\n```python\nlist = [\n {\"name\": \"Node A\"},\n {\"name\": \"Node B\"},\n]\n# Short cut for map(lambda x: Node(**x))\nnodes = Stream(list).map_kwargs(Node)\n```\n\n### Stream many\n\n```python\nmany = Stream.of_many([1, 2, 3], (4, 5, 6))\nmany = many.concat([7, 8, 9])\n```\n\n### End of stream\n```python\nstream = Stream([\"a\", \"b\", \"c\"]).on_end(lambda: print(\"Finished\"))\nchar = stream.next().get()\nif char == \"a\":\n stream.end()\n```\n\n## Examples\n\nA given data structure like:\n```python\ndata = {\n \"ranges\": [\n {\"days\": 3},\n {\"weeks\": 1},\n ]\n}\n```\n\nWithout tinystream:\n```python\nif \"ranges\" in data:\n range_data: timedelta\n for range_data in map(lambda x: timedelta(**x), data[\"ranges\"]):\n pass\n```\n\nWith tinystream:\n```python\nfor range_data in Opt(data).map_key(\"ranges\").stream().map_kwargs(timedelta):\n pass\n```\n\n## Comparison with other libraries\n\nThere are a couple of other implementation to fulfill similar requirements.\n\n- https://github.com/vxgmichel/aiostream\n- https://github.com/python-streamz/streamz\n- https://pypi.org/project/fluentpy\n- https://github.com/ramsteak/streams \n- https://github.com/alemazzo/Python-Java-Stream (*outdated*)\n- https://github.com/JaviOverflow/python-streams (*outdated*)\n- https://github.com/9seconds/streams/ (*outdated*)\n- https://github.com/tolsac/streampy (*outdated*)\n- Apache Spark\n\n## Run the tests\n\n```shell\nPYTHONPATH=\".\" pytest --cov=tinystream -n 4 tests/\n```\n\n### Release update\n1. Update version in `setup.py`\n2. Package library\n ```shell\n python setup.py sdist\n ```\n3. Publish library\n ```shell\n twine upload dist/tinystream-[version].tar.gz\n ```\n\n## References\n\n- https://github.com/MichaelKim0407/tutorial-pip-package\n- https://packaging.python.org/en/latest/guides/making-a-pypi-friendly-readme/\n",
"bugtrack_url": null,
"license": null,
"summary": "Yet another python streams library",
"version": "0.1.15",
"project_urls": {
"Homepage": "https://github.com/mreiche/python-streams"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a16e94cc00b45af7f3ea28122fa444fd33b94823e2bb23e9f392fbc99d5374b6",
"md5": "19e0a653749d9e3f883196710d3b8c37",
"sha256": "063684dbb6eae1dd0c2b38bcb35fd067d96f23a98943b5928a3e45b0aa86ac03"
},
"downloads": -1,
"filename": "tinystream-0.1.15.tar.gz",
"has_sig": false,
"md5_digest": "19e0a653749d9e3f883196710d3b8c37",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 5885,
"upload_time": "2024-03-26T16:40:03",
"upload_time_iso_8601": "2024-03-26T16:40:03.576779Z",
"url": "https://files.pythonhosted.org/packages/a1/6e/94cc00b45af7f3ea28122fa444fd33b94823e2bb23e9f392fbc99d5374b6/tinystream-0.1.15.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-03-26 16:40:03",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "mreiche",
"github_project": "python-streams",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "tinystream"
}
JSON
