Available at [PyPi](https://pypi.org/project/iterators/)
Provides a wrapper class TimeoutIterator to add timeout feature to normal iterators
### Installation:
pip install iterators
See help of TimeoutIterator for all the features. Check tests for examples on how to use TimeoutIterator.
See example tests below for basic usage
### Example:
1. TimeoutIterator works like normal iterator:
```
from iterators import TimeoutIterator
def iter_simple():
yield 1
yield 2
def test_normal_iteration(self):
i = iter_simple()
it = TimeoutIterator(i)
self.assertEqual(next(it), 1)
self.assertEqual(next(it), 2)
self.assertRaises(StopIteration, next, it)
self.assertRaises(StopIteration, next, it)
```
1. When timeout is needed, use like this
```
def iter_with_sleep():
yield 1
time.sleep(0.6)
yield 2
time.sleep(0.4)
yield 3
def test_fixed_timeout(self):
i = iter_with_sleep()
it = TimeoutIterator(i, timeout=0.5)
self.assertEqual(next(it), 1)
self.assertEqual(next(it), it.get_sentinel())
self.assertEqual(next(it), 2)
self.assertEqual(next(it), 3)
self.assertRaises(StopIteration, next, it)
```
1. Dynamic timeout adjustment
```
def iter_with_sleep():
yield 1
time.sleep(0.6)
yield 2
time.sleep(0.4)
yield 3
def test_timeout_update(self):
i = iter_with_sleep()
it = TimeoutIterator(i, timeout=0.5)
self.assertEqual(next(it), 1)
self.assertEqual(next(it), it.get_sentinel())
it.set_timeout(0.3)
self.assertEqual(next(it), 2)
self.assertEqual(next(it), it.get_sentinel())
self.assertEqual(next(it), 3)
self.assertRaises(StopIteration, next, it)
```
### Run unit tests locally:
python -m unittest discover tests
Raw data
{
"_id": null,
"home_page": "https://github.com/leangaurav/pypi_iterator",
"name": "iterators",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.6",
"maintainer_email": "",
"keywords": "ITERATOR,TIMEOUT,SYNC",
"author": "leangaurav",
"author_email": "leangaurav.me@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/02/c4/135b5bdb9f14f728fe1274361b336f77c5f1606af9a5622a765fe75f5fa0/iterators-0.2.0.tar.gz",
"platform": null,
"description": "\n\nAvailable at [PyPi](https://pypi.org/project/iterators/)\n\nProvides a wrapper class TimeoutIterator to add timeout feature to normal iterators\n\n### Installation:\n\n pip install iterators\n\n\nSee help of TimeoutIterator for all the features. Check tests for examples on how to use TimeoutIterator.\nSee example tests below for basic usage\n\n### Example:\n\n1. TimeoutIterator works like normal iterator:\n\n ```\n from iterators import TimeoutIterator\n\n def iter_simple():\n yield 1\n yield 2\n\n def test_normal_iteration(self):\n i = iter_simple()\n it = TimeoutIterator(i)\n\n self.assertEqual(next(it), 1)\n self.assertEqual(next(it), 2)\n\n self.assertRaises(StopIteration, next, it)\n self.assertRaises(StopIteration, next, it)\n ```\n\n1. When timeout is needed, use like this\n ``` \n def iter_with_sleep():\n yield 1\n time.sleep(0.6)\n yield 2\n time.sleep(0.4)\n yield 3\n\n def test_fixed_timeout(self):\n i = iter_with_sleep()\n it = TimeoutIterator(i, timeout=0.5)\n self.assertEqual(next(it), 1)\n self.assertEqual(next(it), it.get_sentinel())\n \n self.assertEqual(next(it), 2)\n self.assertEqual(next(it), 3)\n self.assertRaises(StopIteration, next, it)\n ```\n\n1. Dynamic timeout adjustment\n ```\n def iter_with_sleep():\n yield 1\n time.sleep(0.6)\n yield 2\n time.sleep(0.4)\n yield 3\n\n def test_timeout_update(self):\n i = iter_with_sleep()\n it = TimeoutIterator(i, timeout=0.5)\n self.assertEqual(next(it), 1)\n self.assertEqual(next(it), it.get_sentinel())\n \n it.set_timeout(0.3)\n self.assertEqual(next(it), 2)\n self.assertEqual(next(it), it.get_sentinel())\n\n self.assertEqual(next(it), 3)\n self.assertRaises(StopIteration, next, it)\n ```\n\n### Run unit tests locally:\n python -m unittest discover tests\n",
"bugtrack_url": null,
"license": "",
"summary": "Iterator utility classes and functions",
"version": "0.2.0",
"split_keywords": [
"iterator",
"timeout",
"sync"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "bda19c29772ac9f3bdf9837c92ba5c1fc93f75da14c2e0c3fc41e10485f68feb",
"md5": "3214589300b6790851943174f4eaf292",
"sha256": "1d7ff03f576c9de0e01bac66209556c066d6b1fc45583a99cfc9f4645be7900e"
},
"downloads": -1,
"filename": "iterators-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3214589300b6790851943174f4eaf292",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.6",
"size": 5022,
"upload_time": "2023-01-23T16:07:00",
"upload_time_iso_8601": "2023-01-23T16:07:00.352986Z",
"url": "https://files.pythonhosted.org/packages/bd/a1/9c29772ac9f3bdf9837c92ba5c1fc93f75da14c2e0c3fc41e10485f68feb/iterators-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "02c4135b5bdb9f14f728fe1274361b336f77c5f1606af9a5622a765fe75f5fa0",
"md5": "7b0bb7e8281202cb5f5e44462c4d37ce",
"sha256": "e9927a1ea1ef081830fd1512f3916857c36bd4b37272819a6cd29d0f44431b97"
},
"downloads": -1,
"filename": "iterators-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "7b0bb7e8281202cb5f5e44462c4d37ce",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.6",
"size": 4284,
"upload_time": "2023-01-23T16:07:02",
"upload_time_iso_8601": "2023-01-23T16:07:02.460600Z",
"url": "https://files.pythonhosted.org/packages/02/c4/135b5bdb9f14f728fe1274361b336f77c5f1606af9a5622a765fe75f5fa0/iterators-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-01-23 16:07:02",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "leangaurav",
"github_project": "pypi_iterator",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "iterators"
}
JSON
