Metadata-Version: 2.1
Name: newpool
Version: 0.3.14
Summary: 更简单的使用异步协程和线程池 / 进程池
Author: jiaosenvip
Author-Email: jiaosenvip <jiaosenvip@163.com>
License: MIT
Project-URL: Homepage, https://github.com/jiaosenvip/newpool
Project-URL: Documentation, https://github.com/jiaosenvip/newpool/wiki
Requires-Python: >=3.11
Requires-Dist: rich
Requires-Dist: loky>=3.5.6
Description-Content-Type: text/markdown

# newpool

一个简化异步协程、线程池和进程池使用方式的 Python 工具库。

## 特性

- 基于 `asyncio` 的协程并发：`async_gather()`、`async_taskgroup()`；
- 批量线程执行：`thread_map()`；
- 批量进程执行：`process_map()`；
- 统一的 `Pool` API，可选择线程池或 `loky` 进程池；
- 将同步函数包装为异步函数：`to_async()`；
- 支持任务状态查询、异常统计和已完成任务回收。

## 安装

```bash
pip install -U newpool
```

- Python：`>=3.11`；
- 运行依赖：`rich`、`loky`。

## 快速选择

| 场景 | 推荐 API |
| --- | --- |
| 已经是协程函数 | `async_gather()` 或 `async_taskgroup()` |
| I/O 密集型同步函数 | `thread_map()` 或线程模式 `Pool` |
| CPU 密集型函数 | `process_map()` 或进程模式 `Pool` |
| 单个同步函数需要在协程中执行 | `to_async()` |

## 协程并发

### `async_gather()`

并发执行协程，返回与输入顺序一致的结果列表。单个任务发生异常时不会自动中断其他任务，异常对象会作为对应位置的结果返回。

```python
import asyncio
import newpool


async def work(value):
    await newpool.async_sleep(0.1)
    return value * 2


async def main():
    results = await newpool.async_gather(
        [work(i) for i in range(20)],
        sem=5,  # 同时运行的协程数
    )
    print(results)


asyncio.run(main())
```

### `async_taskgroup()`

基于 `asyncio.TaskGroup` 并发执行协程，返回已创建的 Task 列表。任务发生异常时，TaskGroup 会按 `asyncio` 语义取消其他未完成任务并抛出异常组。

```python
import asyncio
import newpool


async def work(value):
    await newpool.async_sleep(0.1)
    return value * 2


async def main():
    tasks = await newpool.async_taskgroup(
        [work(i) for i in range(20)],
        sem=5,
    )
    print([task.result() for task in tasks])


asyncio.run(main())
```

参数：

- `coros_or_futures`：单个协程或协程列表；
- `sem`：并发限制。省略或传 `None` 时不使用信号量限制。

## 线程池

### 批量执行：`thread_map()`

```python
import newpool


def square(value):
    return value * value


results = newpool.thread_map(square, range(10), n=4)
print(results)
```

参数：

- `n`：线程数。省略时使用 `min(32, CPU 核心数 + 4)`；
- `timeout`：等待映射结果的超时时间；
- `chunksize`：兼容统一 map 接口的参数。线程池通常不需要调整。

### 长生命周期线程池：`Pool()`

```python
import time
import newpool


pool = newpool.Pool(n=5)


@pool.task
def work(value):
    time.sleep(0.1)
    return value * 2


try:
    futures = [work(i) for i in range(20)]
    pool.wait()
    print([future.result() for future in futures])
    print(pool.state_str())
finally:
    pool.shutdown()
```

线程池中的任务通过 `Future` 返回结果和异常：

```python
future = pool.run(work, 1)
pool.wait()

try:
    print(future.result())
except Exception as exc:
    print("任务失败:", exc)
```

长期运行或需要隔离任务时，建议自己创建 `Pool`，这样可以独立调用 `shutdown()`。库也提供了一个内置共享线程池：

```python
import newpool


@newpool.pool.task
def work(value):
    return value * 2


futures = [work(i) for i in range(10)]
newpool.pool.wait()
print([future.result() for future in futures])
```

`newpool.pool` 会在模块导入时创建，并由调用方负责管理；如果调用了 `newpool.pool.shutdown()` 或 `newpool.pool.kill()`，这个共享池就不能继续提交任务，需要重新创建独立的 `Pool`。

## 进程池

进程池基于 `loky`，适合 CPU 密集型任务。任务函数和参数需要能够被序列化。Windows 或使用进程启动方式的环境中，请把启动代码放在 `if __name__ == "__main__":` 保护下。

### 批量执行：`process_map()`

```python
import newpool


def square(value):
    return value * value


if __name__ == "__main__":
    results = newpool.process_map(square, range(10), n=4)
    print(results)
```

### 长生命周期进程池：`Pool(mode="process")`

```python
import newpool


def work(value):
    return value * value


if __name__ == "__main__":
    pool = newpool.Pool(n=4, mode="process")
    try:
        futures = [pool.run(work, i) for i in range(10)]
        pool.wait()
        print([future.result() for future in futures])
        print(pool.state_dict())
    finally:
        pool.shutdown()
```

也可以使用装饰器：

```python
import newpool


if __name__ == "__main__":
    pool = newpool.Pool(n=4, mode="process")
    try:
        @pool.task
        def work(value):
            return value * value

        futures = [work(i) for i in range(10)]
        pool.wait()
        print([future.result() for future in futures])
    finally:
        pool.shutdown()
```

注意：进程池任务异常不会在提交任务时直接抛出。请通过 `future.result()` 读取结果，或通过 `future.exception()` 检查异常。不要把连接、锁、打开的文件句柄等不可序列化对象作为进程任务参数。

## 将同步函数包装为异步：`to_async()`

`to_async()` 使用线程池执行同步函数，不会把同步函数变成真正的协程函数。位置参数和关键字参数都支持：

```python
import asyncio
import newpool


@newpool.to_async()
def add(a, b=0):
    return a + b


async def main():
    print(await add(2, 3))
    print(await add(2, b=3))
    print(await add(a=2, b=3))


asyncio.run(main())
```

可以通过 `n` 指定线程数：

```python
@newpool.to_async(n=5)
def work(value):
    return value * 2
```

长期运行程序中，建议注意显式线程池的生命周期；`to_async(n=5)` 创建的线程池没有单独的公开关闭接口。

## `Pool` 状态、清理和关闭

### `wait()`

等待当前池中跟踪的任务完成。默认等待全部任务完成：

```python
pool.wait()
```

`Pool` 支持动态提交任务，等待期间追加到 `results` 的任务可能会被继续纳入等待。当前版本的 `timeout` 和 `return_when` 参数不建议作为复杂控制流程的唯一依据；需要精确超时控制时，请直接使用对应的 `Future` 或标准库等待接口。

### `state_dict()`、`state_str()`、`state()`

```python
state = pool.state_dict()  # 字典
text = pool.state_str()    # 多行字符串
pool.state()               # Rich Panel 输出
```

状态分为累计统计和当前任务状态：

- `总任务`、`已完成`、`成功`、`取消`、`报错`：累计统计，包含已经归档和当前仍保留的 Future；
- `已归档`：已经通过 `clear_done()` 清理的 Future 数量；
- `当前任务`：仍保留在 `results` 中的 Future 数量；
- `当前已完成`：已完成但尚未调用 `clear_done()` 的 Future 数量；
- `正在运行`：当前正在执行的任务数量；
- `剩余任务`：已经提交但尚未开始执行的任务数量。

### `clear_done()`

归档并清理已经完成的 Future，保留正在运行和等待中的任务。归档只保存数字，不保存 Future 或结果对象，适合大批量或长期运行的任务。

```python
summary = pool.clear_done()
```

调用时会直接输出一条摘要，例如：

```text
已归档 10 个已完成任务：成功 8,报错 1,取消 1
```

同时返回：

```python
{
    "已归档": 10,
    "成功": 8,
    "报错": 1,
    "取消": 1,
}
```

没有可清理任务时会输出：

```text
没有可归档的已完成任务
```

调用方如果还需要结果，应在清理前读取 `Future.result()`：

```python
for future in pool.results:
    if future.done():
        try:
            save_result(future.result())
        except Exception as exc:
            save_error(exc)

pool.clear_done()
```

### `clear()`

彻底清空当前任务和统计归档：

- 尝试取消尚未开始的任务；
- 清空 `results`；
- 将总任务、成功、报错和取消统计归零。

已经开始执行的线程任务不一定能够被强制取消，调用后底层线程可能继续运行，但不会再被池跟踪。只想清理已完成任务并保留累计统计时，应使用 `clear_done()`。

### `cancel_all()`、`shutdown()` 和 `kill()`

- `cancel_all()`：尝试取消所有尚未开始的任务；正在运行的任务通常不能被取消；
- `shutdown(wait=True, cancel_futures=False)`：关闭池。传入 `cancel_futures=True` 时，会先调用 `cancel_all()`；进程池还会终止 worker；
- `kill()`：等价于 `shutdown(wait=False, cancel_futures=True)`。进程池会终止 worker；线程池只能取消尚未开始的任务，不能安全强杀正在运行的线程。

关闭后不要继续向同一个池提交任务；如果还要执行新任务，请重新创建 `Pool`。

## API 概览

```text
async_gather(coros_or_futures, sem=None)
async_taskgroup(coros_or_futures, sem=None)
asyncio_run(coro)
thread_map(func, *iterables, n=None, timeout=None, chunksize=1)
process_map(func, *iterables, n=None, timeout=None, chunksize=1)
to_async(n=None)

Pool(n=None, mode="thread")
Pool.run(func, *args, **kwargs)
Pool.task(func)
Pool.callback(func)
Pool.wait(timeout=None, return_when="ALL_COMPLETED")
Pool.cancel_all()
Pool.clear_done()
Pool.clear()
Pool.state_dict()
Pool.state_str(extend=None)
Pool.state(extend=None)
Pool.shutdown(wait=True, cancel_futures=False)
Pool.kill()
```

## 许可证

MIT License。
