vxutils 20260127
A toolbox for vxquant
License: MIT
Requires Python: >=3.10
Keywords: cache, convertors, database, logger, tools, utils

# vxutils

**vxutils** 是一个为 Python 开发提供常用工具的集合库，包含日志配置、数据转换、缓存管理、装饰器工具、数据库封装以及线程池等实用功能。它旨在简化日常开发任务，提供开箱即用的解决方案。

## 功能特性

*   **Logger**: 强大的日志配置工具，支持控制台彩色输出、文件轮转、异步日志记录。
*   **Convertors**: 丰富的数据类型转换工具，涵盖时间、JSON、枚举等常见类型。
*   **Decorators**: 实用的装饰器集合，包括重试、计时、单例、超时控制、限流等。
*   **Cache**: 基于 SQLite 的 TTL 缓存管理器，支持 Python 原生对象及 Pandas/Polars DataFrame 的持久化缓存。
*   **Database**: 轻量级 SQLite 数据库封装，支持对象关系映射（ORM）风格的操作。
*   **Executor**: 增强型线程池执行器，支持自动回收空闲线程，节省资源。
*   **Tools**: 其他实用工具，如 API Key 管理。

## 安装

```bash
pip install vxutils
```

或者使用 `uv`:

```bash
uv add vxutils
```

## 模块详解与示例

### 1. 日志工具 (Logger)

提供了一键配置日志的功能，支持彩色输出和异步写入。

```python
from vxutils import loggerConfig
import logging

# 配置日志
logger = loggerConfig(
    level="DEBUG",
    colored=True,             # 开启控制台彩色输出
    filename="logs/app.log",  # 日志文件路径
    async_logger=True,        # 开启异步日志，不阻塞主线程
    when="D",                 # 按天轮转
    backup_count=7            # 保留7天日志
)

logger.info("这是一条普通信息")
logger.error("这是一条错误信息")
```

### 2. 装饰器 (Decorators)

包含多种增强函数功能的装饰器。

```python
from vxutils import retry, timer, timeout, rate_limit, singleton
import time

# 1. 重试装饰器
@retry(max_retries=3, delay=1)
def unstable_network_call():
    print("尝试连接...")
    raise ConnectionError("连接失败")

# 2. 计时装饰器
@timer(descriptions="耗时操作", verbose=True)
def heavy_computation():
    time.sleep(0.5)

# 3. 超时装饰器
@timeout(seconds=1.0)
def long_running_task():
    time.sleep(2.0)  # 将抛出 TimeoutError

# 4. 限流装饰器 (例如：1秒内最多调用2次)
@rate_limit(times=2, period=1.0)
def api_request():
    pass

# 5. 单例装饰器
@singleton
class DatabaseConnection:
    pass
```

### 3. 数据转换 (Convertors)

统一的时间和数据格式转换接口。

```python
from vxutils import to_datetime, to_timestring, to_json
import datetime

# 时间转换
dt = to_datetime("2023-01-01 12:00:00")
ts_str = to_timestring(datetime.datetime.now())

# JSON 转换 (支持 datetime, Enum 等特殊类型)
data = {
    "now": datetime.datetime.now(),
    "status": "active"
}
json_str = to_json(data)
```

### 4. 数据库封装 (Database)

轻量级的 SQLite 操作封装，支持链式调用和简单的 ORM。

```python
from vxutils import SQLiteConnectionWrapper, SQLExpr

# 连接数据库 (支持 :memory: 或文件路径)
with SQLiteConnectionWrapper("my_data.db") as conn:
    # 创建表 (通常配合 register 使用，或直接 execute)
    conn.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)")
    
    # 插入/更新数据 (支持 UPSERT)
    conn.save("users", {"id": 1, "name": "Alice", "age": 30})
    
    # 查询单条
    user = conn.findone("users", SQLExpr("name") == "Alice")
    print(user.name, user.age)
    
    # 复杂查询
    expr = (SQLExpr("age") > 20) & (SQLExpr("name").like("A%"))
    users = conn.find("users", expr).fetchall()
    
    # 统计
    count = conn.count("users", SQLExpr("age") > 25)
```

### 5. 缓存管理 (Cache)

基于 SQLite 的本地持久化缓存，特别优化了对 DataFrame 的支持。

```python
from vxutils import Cache
import pandas as pd

cache = Cache("my_cache.db")

# 缓存普通数据
cache.set("my_key", {"a": 1, "b": 2}, ttl=60) # 60秒过期

# 缓存 DataFrame
df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]})
cache.set("df_key", df, ttl=3600)

# 获取数据
data = cache.get(my_key_param="val") # 支持根据参数生成 key
```

### 6. 线程池 (Executor)

`VXThreadPoolExecutor` 是 `ThreadPoolExecutor` 的增强版，支持空闲线程自动回收。

```python
from vxutils import VXThreadPoolExecutor
import time

# 创建一个最大 10 线程，空闲 5 秒后回收线程的线程池
with VXThreadPoolExecutor(max_workers=10, idle_timeout=5.0) as executor:
    future = executor.submit(time.sleep, 1)
    print(future.result())
```

### 7. 其他工具 (Tools)

**APIKeyManager**: 简单的本地 API Key 管理工具，避免在代码中硬编码密钥。

```python
from vxutils import APIKeyManager

km = APIKeyManager("./secrets.json")
# 如果文件中不存在该 key，会提示用户输入并保存
api_key = km.get_key("OPENAI_API_KEY")
```

## 开发与测试

本项目使用 `pytest` 进行测试。

```bash
# 安装开发依赖
uv sync --dev

# 运行测试
pytest tests/
```

## License

MIT License