vxutils 20260225

A toolbox for vxquant

vxutils

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

功能特性

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

安装

pip install vxutils

或者使用 uv:

uv add vxutils

模块详解与示例

1. 日志工具 (Logger)

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

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)

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

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)

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

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。

同步模式

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)
    # find 返回 SQLiteRow 列表，无需调用 fetchall()
    for user in users:
        print(user.name)
    
    # 统计
    count = conn.count("users", SQLExpr("age") > 25)

异步模式 (Async)

需要安装 aiosqlite：pip install vxutils[async]

import asyncio
from vxutils import AsyncSQLiteConnectionWrapper, SQLExpr

async def main():
    async with AsyncSQLiteConnectionWrapper("my_async.db") as db:
        # 注册表结构
        # await db.register("users", UserClass)
        
        # 插入数据
        await db.save("users", {"name": "Bob", "age": 25})
        
        # 查询
        rows = await db.find("users", SQLExpr("age") > 20)
        for row in rows:
            print(row)

if __name__ == "__main__":
    asyncio.run(main())

5. 缓存管理 (Cache)

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

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 的增强版，支持空闲线程自动回收。

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 管理工具，避免在代码中硬编码密钥。

from vxutils import APIKeyManager

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

开发与测试

本项目使用 pytest 进行测试。

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

# 运行测试
pytest tests/

License

MIT License