Server IP : 192.64.118.117 / Your IP : 18.224.2.133 Web Server : LiteSpeed System : Linux premium56.web-hosting.com 4.18.0-513.24.1.lve.1.el8.x86_64 #1 SMP Thu May 9 15:10:09 UTC 2024 x86_64 User : thecgapy ( 1160) PHP Version : 7.4.33 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/utils/ |
Upload File : |
from typing import List, Union, Tuple import asyncio class AsyncIterate: # not AsyncIterable because python use this name already def __init__(self, data: Union[List, Tuple]): self.queue = iter(data) def __aiter__(self): return self async def __anext__(self): data = await self.fetch_data() if data is not None: return data else: raise StopAsyncIteration async def fetch_data(self): try: item = next(self.queue) except StopIteration: item = None return item async def gather(*tasks: List) -> AsyncIterate: results = await asyncio.gather(*tasks) return AsyncIterate(results)