yayo-pypkg 0.0.8

yayo 工具包

yayo-pypkg

yayo 工具包 —— 提供常用的 Python 工具函数。

Python 版本支持

范围	说明
基础功能 (log / config / utils)	✅ Python >= 3.7,零硬依赖
core extras(常用依赖:HTTP / 雪花 / pydantic / 调度器)	Python >= 3.7
api extras(FastAPI + DB)	Python >= 3.8(由 fastapi 决定)
llm extras(LangChain)	Python >= 3.10(由 langchain 决定)
all extras(全包)	Python >= 3.10

代码兼容 vs 依赖兼容是两回事:

• 代码本身:全部源码用 from __future__ import annotations,PEP 604 (str \| None) / 内建泛型 (list[str]) 都做了惰性化,Python 3.7 就能 import
• 第三方依赖:requests 2.32+ / snowflake-id 1.0+ / langchain 1.x / fastapi 0.100+ 等都要求 Python 3.8+ / 3.10+,装哪个版本由调用方自己决定

安装(按需装)

核心设计:

• pyproject.toml 不钉任何版本上下限 —— 包名只声明"我用到这个",具体版本由你控制
• 基础安装零三方依赖
• 第三方库全部按需可选,装基础包后用到哪个功能再装哪个

# 基础(零三方依赖,只装 log / config / utils)
pip install yayo-pypkg

# 常用依赖一次装齐(requests + snowflake-id + pydantic + apscheduler)
pip install yayo-pypkg[core]

# FastAPI + SQLAlchemy + MySQL(需要 Python >= 3.8)
pip install yayo-pypkg[api]

# LangChain + DeepSeek / OpenAI(需要 Python >= 3.10)
pip install yayo-pypkg[llm]

# 全包(需要 Python >= 3.10)
pip install yayo-pypkg[all]

uv 用户:把 pip install 换成 uv add 即可,语义完全一样。 也支持装多个 extras:uv add yayo-pypkg[core,api]

想锁定特定版本?(内网老 Python 常用)

# 1. 装基础包(零依赖,Python 3.7 也行)
uv add yayo-pypkg

# 2. 手动装指定版本的依赖(比如 requests 2.31.0 是最后支持 Python 3.7 的)
uv add "requests>=2.28,<2.32"
uv add "snowflake-id>=0.0.5,<1.0"

为什么这样能 work?

• 基础包本身不依赖 requests/snowflake-id(它们在 dependencies=[] 里没有)
• 代码用了 lazy import,没装 requests 时 import yayo_pypkg.http_utils 不报错
• 你装哪个版本都行,只要那个版本本身支持你的 Python 版本就行

缺失依赖时的行为

按需可选的设计带来的好处:没装 extras 也能 import,只有真用到时才报错,而且错误信息会直接告诉你怎么装:

# 没装 [http] 时
from yayo_pypkg import http_utils  # ✅ OK,可以 import
client = http_utils.HttpClient()   # ❌ 抛友好 ImportError:
                                    #     使用 yayo_pypkg.http_utils 需要先安装 requests:
                                    #         uv add yayo-pypkg[http]
                                    #         # 或: pip install 'requests>=2.28,<2.32'

# 没装 [id] 时
from yayo_pypkg import snowflake_utils       # ✅ OK
snowflake_utils.generate_snowflake_id()      # ❌ 抛友好 ImportError:
                                             #     使用 yayo_pypkg.snowflake_utils 需要先安装 snowflake-id:
                                             #         uv add yayo-pypkg[id]

快速上手

支持两种导入方式(完全等价,任选其一):

# 方式 1:完整包名
from yayo_pypkg import hello, config, get_logger

# 方式 2:短别名
from ypk import hello, config, get_logger

print(hello())         # -> hello from yayo_pypkg
print(config.DB_HOST)  # -> 从 .env 自动读取(支持类型自动转换)

日志(get_logger)

一行接入,按天切割 + 保留 7 天 + 控制台/文件双输出:

# log.py
from yayo_pypkg import get_logger
logger = get_logger()     # 自动用入口脚本名,如 main.py → ./logs/main.log
logger.info("hello")
logger.error("oops")

• 默认自动在 CWD 下创建 ./logs/<name>.log
• 同名 logger 幂等,多次 get_logger() 不会重复挂 handler
• 业务方只需 from .log import logger 即可

配置(config)

读取项目根目录的 .env,支持 str / int / float / bool 自动类型推断:

# .env
DB_HOST=localhost
DB_PORT=3306
DEBUG=true
APP_NAME="my service"

from yayo_pypkg import config

print(config.DB_HOST)    # -> 'localhost'  (str)
print(config.DB_PORT)    # -> 3306         (int,自动转换)
print(config.DEBUG)      # -> True         (bool,自动转换)
print(config.APP_NAME)   # -> 'my service' (引号自动剥离)

• 环境变量优先级高于 .env 文件
• 访问不存在的属性会抛 AttributeError,IDE 友好

Redis(yayo_pypkg.db.db_redis)

异步 Redis 客户端,基于 redis.asyncio(redis>=4.2),可选依赖(在 [api] extras 里)。自动支持单节点和 3+ 节点集群。

.env 配置 —— 单节点:

REDIS_ENABLE=true
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=
REDIS_TIMEOUT=10

.env 配置 —— 3 节点集群(自动检测:REDIS_CLUSTER_NODES 非空就走集群模式):

REDIS_ENABLE=true
REDIS_CLUSTER_NODES=192.168.1.10:6379,192.168.1.11:6379,192.168.1.12:6379
REDIS_PASSWORD=cluster_secret       # 集群统一密码
REDIS_TIMEOUT=10
# REDIS_HOST / REDIS_PORT / REDIS_DB 在集群模式下被忽略

基础用法(单节点 / 集群用法完全一样,不用改业务代码):

import asyncio
from yayo_pypkg.db.db_redis import (
    init_redis, close_redis,
    set_value, get_value,
    set_json, get_json,
    delete, exists, expire, mget,
)

async def main():
    await init_redis()                          # 自动选模式 + 读 .env
    try:
        # 基础 KV
        await set_value("foo", "bar")
        print(await get_value("foo"))           # 'bar'

        # 带 TTL
        await set_value("session:abc", "xyz", ttl=3600)

        # JSON 快捷(自动序列化,中文不转义)
        await set_json("user:1", {"name": "张三", "age": 18}, ttl=3600)
        print(await get_json("user:1"))         # {'name': '张三', 'age': 18}

        # 批量操作
        vals = await mget("a", "b", "c")        # 批量读,不存在的位置是 None
        n = await delete("a", "b")              # 批量删
        n = await exists("a", "b", "c")         # 存在几个
        await expire("foo", 60)                 # 设置过期时间(秒)
    finally:
        await close_redis()

asyncio.run(main())

FastAPI lifespan 集成:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from yayo_pypkg.db.db_redis import init_redis, close_redis, get_redis

@asynccontextmanager
async def lifespan(app):
    await init_redis()                  # 自动选单节点 / 集群
    yield
    await close_redis()

app = FastAPI(lifespan=lifespan)

@app.get("/cache/{key}")
async def get_cache(key: str):
    r = get_redis()                # 拿到原始 redis 客户端(单节点 or RedisCluster)
    val = await r.get(key)         # 所有 redis-py 原生 API 都能用(pipeline / pubsub / hash / list / zset 都有)
    return {"value": val.decode() if val else None}

集群 vs 单节点的差别(对业务代码透明):

维度	单节点 Redis	集群 RedisCluster
启动参数	host + port + db	startup_nodes=[{host,port}, ...]
key 路由	直接定位	客户端按 hash slot 自动路由
多 key 操作限制	无	所有 key 必须在同一 hash slot(用 {tag} 强制同 slot)
事务 / pipeline	支持	仅同 slot 的 key 才行
init_redis() 走哪条	REDIS_CLUSTER_NODES 为空	REDIS_CLUSTER_NODES 非空

集群注意事项(业务代码需要知道):

• 如果用 mget("a", "b", "c") / delete("a", "b") / MSET 等多 key 命令,所有 key 必须在同一 hash slot,否则报 CROSSSLOT
• 想强制一组 key 落到同 slot,用 hashtag:user:{1001}.name + user:{1001}.age → {} 里的 1001 是 hash tag,保证这俩 key 同 slot
• 单 key 操作(get / set / expire 等)不受影响

模块概览

模块	说明
yayo_pypkg.core.logger	日志配置(get_logger / setup_logger)
yayo_pypkg.core.config	.env 加载 + 类型推断
yayo_pypkg.cron.scheduler	定时任务调度器([api] extras)
yayo_pypkg.path_utils	路径处理(项目根目录 / 目录创建)
yayo_pypkg.datetime_utils	日期时间工具
yayo_pypkg.str_utils	字符串工具
yayo_pypkg.file_utils	文件读写工具
yayo_pypkg.list_utils	列表/集合工具
yayo_pypkg.json_utils	JSON 工具(pydantic 懒加载,[api] extras)
yayo_pypkg.http_utils	HTTP 客户端封装(基础包自带 requests)
yayo_pypkg.snowflake_utils	雪花 ID 生成器(基础包自带 snowflake-id)
yayo_pypkg.exceptions	自定义异常层级
yayo_pypkg.llm	LLM / DeepSeek 流式服务([llm] extras)
yayo_pypkg.middleware	FastAPI 中间件([api] extras)
yayo_pypkg.schemas	Pydantic Schema 基类([api] extras)
yayo_pypkg.db.db_mysql	MySQL 连接池([api] extras)
yayo_pypkg.db.db_redis	Redis 异步客户端([api] extras,基于 aioredis)
yayo_pypkg.ocr	MinerU OCR 客户端

License

MIT