update

Author: Linux

.github/workflows/mypy_check.yml

@@ -0,0 +1,39 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: mypy
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python 3.10
+      uses: actions/setup-python@v3
+      with:
+        python-version: "3.10"
+
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install .
+        if [ -f requirements-dev.txt ]; then pip install -r requirements-dev.txt; fi
+
+    - name: Lint with ruff
+      run: |
+        ruff check
+        
+    - name: Type checking with mypy
+      run: |
+        mypy --strict --exclude build .

README.md

@@ -0,0 +1,27 @@
+[![ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/charliermarsh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff) 
+[![mypy](https://github.com/guangrei/syncio/actions/workflows/mypy_check.yml/badge.svg)](https://github.com/guangrei/syncio/actions) 
+
+Syncio is inspired by `asyncio`. you can easy to create task and gather with `syncio.gather()` (multiprocessing) or `syncio.thread_gather()` (threading).
+
+## Example
+```python
+from syncio import create_task, gather
+
+
+def hello(n: int) -> str:
+    return f"hello {n + 1}"
+
+
+tasks = [create_task(hello)(i) for i in range(3)]
+results = gather(*tasks)
+print("output task_1:", results["task_1"])
+print("output task_2:", results["task_2"])
+print("output task_3:", results["task_3"])
+
+# or using iterator
+
+for result in results:
+    print(result)
+
+```
+author: guangrei.

requirements-dev.txt

@@ -0,0 +1,6 @@
+mypy
+ruff
+typing-extensions
+twine
+setuptools
+types-setuptools

setup.cfg

@@ -0,0 +1,2 @@
+[bdist_wheel]
+universal = 1

setup.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python
+from setuptools import setup
+from os import path
+
+
+this_directory = path.abspath(path.dirname(__file__))
+with open(path.join(this_directory, "README.md")) as f:
+    long_description = f.read()
+
+requirements = [
+    "typing-extensions",
+]
+
+setup(
+    name="syncio",
+    version="v0.0.1",
+    description="type safe for synchronous python concurrently",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    author="guangrei",
+    author_email="[email protected]",
+    url="https://github.com/guangrei/syncio",
+    packages=["syncio"],
+    package_data={"syncio": ["py.typed"]},
+    license="MIT",
+    platforms="any",
+    install_requires=requirements,
+)

syncio/Multiprocessing.py

@@ -0,0 +1,23 @@
+# -*-coding:utf8;-*-
+from multiprocessing import Process, Manager
+from syncio.Task import NewTask, TaskOutput
+
+
+def gather(*args: NewTask) -> TaskOutput:
+    manager = Manager()
+    ret = manager.dict()
+    tasklist = []
+
+    def runner(tasker: NewTask, task_id: int) -> None:
+        task_id += 1
+        result = tasker.func(*tasker.args, **tasker.kwargs)
+        ret[f"task_{task_id}"] = result
+
+    for k, f in enumerate(args):
+        p = Process(target=runner, kwargs={"tasker": f, "task_id": k})
+        p.start()
+        tasklist.append(p)
+    for p in tasklist:
+        p.join()
+
+    return TaskOutput(dict(ret))

syncio/Task.py

@@ -0,0 +1,51 @@
+# -*-coding:utf8;-*-
+from abc import ABC, abstractmethod
+from typing import Callable, List, Any, Iterator, Dict
+from typing_extensions import Self
+
+
+class NewTask(ABC):
+    func: Callable[..., Any]
+    args: List[Any]
+    kwargs: Dict[str, Any]
+
+    @abstractmethod
+    def __init__(self, func: Callable[..., Any]) -> None:
+        pass
+
+    @abstractmethod
+    def __call__(self, *args: Any, **kwargs: Any) -> Self:
+        pass
+
+
+class create_task(NewTask):
+    def __init__(self, func: Callable[..., Any]) -> None:
+        self.func = func
+        self.args: List[Any] = []
+        self.kwargs: Dict[str, Any] = {}
+
+    def __call__(self, *args: Any, **kwargs: Any) -> Self:
+        self.args = list(args)
+        self.kwargs = dict(kwargs)
+        return self
+
+
+class TaskOutput(dict):  # type: ignore[type-arg]
+    def __init__(self, data: Dict[str, Any]) -> None:
+        self._data = data
+        super().__init__(data)
+
+    def __iter__(self) -> Iterator[Any]:
+        output: List[Any] = []
+        for _, v in self._data.items():
+            output.append(v)
+        self._output_list = output
+        self._index = 0
+        return self
+
+    def __next__(self) -> Any:
+        if self._index >= len(self._output_list):
+            raise StopIteration
+        value: Any = self._output_list[self._index]
+        self._index += 1
+        return value

syncio/Threading.py

@@ -0,0 +1,22 @@
+# -*-coding:utf8;-*-
+from threading import Thread
+from syncio.Task import NewTask, TaskOutput
+
+
+def thread_gather(*args: NewTask) -> TaskOutput:
+    ret = {}
+    tasklist = []
+
+    def runner(tasker: NewTask, task_id: int) -> None:
+        task_id += 1
+        result = tasker.func(*tasker.args, **tasker.kwargs)
+        ret[f"task_{task_id}"] = result
+
+    for k, f in enumerate(args):
+        p = Thread(target=runner, kwargs={"tasker": f, "task_id": k})
+        p.start()
+        tasklist.append(p)
+    for p in tasklist:
+        p.join()
+
+    return TaskOutput(ret)

syncio/__init__.py

@@ -0,0 +1,6 @@
+# -*-coding:utf8;-*-
+from syncio.Task import create_task
+from syncio.Multiprocessing import gather
+from syncio.Threading import thread_gather
+
+__all__ = ["create_task", "gather", "thread_gather"]