semvec/pypi-packages huggingface dataset:
aiotw 0.1
Asynchronous library for working with Twitch.
License: MIT
Requires Python: >=3.11

<h1 align="center">aiotw</h1>
<p align="center">
  aiotw is a small, asynchronous library for working with Twitch.
</p>
<br/>


> [!WARNING]
> PROJECT STILL UNDER DEVELOPMENT! We do not recommend using this project in production BEFORE v1.0! Any update can broke previous functionallity!


# Implemented APIs
- [x] IRC
- [ ] Helix
- [ ] EventSub.


# Requirements
- Python >= 3.11


# Quick start
Install

```bash
pip install aiotw
```

## IRC
Simple client example

```python
import asyncio
from aiotw import irc


client = irc.Client(auth_token='oauth:...', auth_nickname='mybot')
# or you can use anonymous client
client = irc.create_anonymous_client()


@client.on(irc.PrivmsgMessage)
async def on_privmsg(msg: irc.PrivmsgMessage):
    print(f'[{msg.channel}] <{msg.user}>: {msg.message}')

    # send privmsg will not work if you use anonymous client!
    if not client.is_anonymous:
      await client.send(irc.PrivmsgMessage(channel=msg.channel, message=msg.message))


async def main():
    async with client:
        await client.send(irc.JoinMessage(channel='some_channel'))
        await client.serve()


asyncio.run(main())
```

_See [irc sources](./src/aiotw/irc.py) for details._

## Helix
_Not implemented yet._

## EventPub
_Not implemented yet._

# License
This project is released under the MIT License — see the [LICENSE file](./LICENSE) for the full text.
newpool 0.2.0
更简单的使用异步协程和线程池
License: MIT
Requires Python: >=3.11

# newpool

>更简单的使用异步协程和线程池

## 安装

```
pip install newpool
```

## 使用

### 异步协程

`newpool.async_gather()` 并发执行多个协程, 返回结果列表, 遇到异常不会中断, 而是将错误原因返回到结果列表中
`newpool.async_taskgroup()` 并发执行多个协程, 返回列表, 包含更详细的状态, 遇到异常会中断

```py
import newpool

async def test():
    await newpool.async_sleep(1)
    print("test")

gather_results = newpool.async_gather([test() for i in range(20)], sem=5) # sem 为并发协程数
taskgroup_results = newpool.async_taskgroup([test() for i in range(20)]) # 此处没有使用 sem, 默认基于设备自动计算并发协程数
```

### 线程池

**基于 `concurrent.futures.ThreadPoolExecutor` 实现**

```py
import newpool
import time

pool = newpool.Pool(5)

@pool.task
def test():
    time.sleep(2)
    print("test")

for i in range(20):
    test()

pool.wait() # 等待所有任务执行完毕
pool.state() # 线程池状态
pool.results # 所有任务执行结果
pool.cancel_all() # 取消所有任务
```

内置可以直接使用的线程池 `newpool.pool`, 无需实例化。
线程数默认基于设备自动计算, 也可以通过 `newpool.pool.n = 5` 修改线程数 

```py
import newpool

@newpool.pool.task
def test():
    print("test")

for i in range(20):
    test()

newpool.pool.wait() # 等待所有任务执行完毕
newpool.pool.state() # 线程池状态
newpool.pool.results # 所有任务执行结果
newpool.pool.cancel_all() # 取消所有任务
```

**基于 `asyncio` 实现的线程池**

使用 `newpool.async_gather()` 或者 `newpool.async_taskgroup()`, 通过 `asyncio.run` 运行。

默认线程数基于设备自动计算，也可以通过 `newpool.to_async(n=5)` 修改线程数。

```py
import newpool
import time

@newpool.to_async()
def test():
    time.sleep(2)
    print("test")

tasks = newpool.async_gather([test() for i in range(20)])
newpool.asyncio_run(tasks)

@newpool.to_async(n=5)
def test2():
    time.sleep(2)
    print("test")

tasks2 = newpool.async_gather([test2() for i in range(20)])
newpool.asyncio_run(tasks2)
```