yayo-pypkg

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

Python 版本支持

代码兼容 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]

# PyMuPDF / fitz(PDF → 图片,1.x 需要 Python >= 3.8)
pip install yayo-pypkg[image]

# 全包(需要 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]

快速上手

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

# 完整包名
from yayo_pypkg import hello, config, get_logger

# 或用短别名(推荐,少打几个字)
from ypk import hello, config, get_logger

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

关于 ypk 别名：ypk 是 yayo_pypkg 的短别名包,两者完全等价(ypk.logger is yayo_pypkg.logger)。 PyCharm 静态分析有时会误报 from ypk.xxx import yyy,运行完全正常;如果在意,直接用 yayo_pypkg。

命令行工具(ypkstart / ypkstop / ypkstatus)

装了 yayo-pypkg 之后,会在系统 PATH 里装上 5 个命令,可以在任何 FastAPI 项目根目录直接用:

cd /path/to/your-fastapi-project   # 有 main.py + .env 的目录

ypkstart                          # 启动服务(后台,默认 supervisor 守护)
ypkstatus                         # 查看状态
ypkstop                           # 停止服务
ypklog                            # tail -f 日志(Ctrl+C 退出)
ypk start --reload                # 开发模式(代码改动自动 reload)
ypk start --env production        # 加载 .env.production 多环境配置
ypk start --no-supervisor         # 关闭 supervisor,回到旧行为(子进程死了不自动拉起)

自动做的事

ypkstart 会自动:

1. 在 CWD 找 .env,找不到在上级目录找(适合 monorepo)
2. 支持多环境: --env production 加载 .env.production 并回退到 .env(也可用 APP_ENV 环境变量)
3. 解析 .env 加载到 os.environ(已有变量不覆盖,shell 环境变量优先级最高)
4. 找入口模块: main.py / app.py / server.py(按顺序)
5. 从 .env 读 PORT 或 YPK_PORT(host 同理)
6. 端口预检: 被占时给出明确错误,--clean 可强杀端口上的进程
7. preflight: 提前 import 业务模块,捕获依赖缺失(比"uvicorn 秒挂"快 ~1s)
8. 后台启动 uvicorn,日志写 .ypk.log,PID 写 .ypk.pid
9. supervisor 守护 (默认启用):
   • 子进程被 OOM Killer 杀掉 → 自动重启
   • RSS 逼近内存上限 → 主动 SIGTERM 触发 graceful restart(避免 OOM)
   • 短窗口内重启过多 → 自动放弃(防 OOM 雪崩)
   • 加 --no-supervisor 可关闭

项目目录长这样就能直接用

your-project/
├── .env                  # 包含 PORT=8080 等
├── main.py               # 你的 FastAPI app
├── .ypk.pid              # ypk 自动生成(PID)
└── .ypk.log              # ypk 自动生成(日志)

完整示例

$ cd ~/projects/myapp
$ ypkstart
📄 .env (env=default): ['.env'] (共新加载 5 个变量)
📦 模块: main:app
🌐 监听: 0.0.0.0:8080
🚀 启动命令: /path/to/python -m uvicorn main:app --host 0.0.0.0 --port 8080
🛡️  Supervisor 模式启用 (子进程被 OOM 杀会自动重启)
   内存阈值: WARN=80% HARD=95%
   重启策略: 5 次 / 300.0s 内
   关闭:    加 --no-supervisor
   端口:    8080
👀 Supervisor 启动 (PID=12345)
📋 命令: /path/to/python -m uvicorn main:app ...
🚀 启动子进程 PID=12346

$ curl http://localhost:8080
{"hello": "world"}

$ ypkstatus
✅ 正在跑 PID=12345
   端口: 0.0.0.0:8080
   日志: /home/me/projects/myapp/.ypk.log
   PID 文件: /home/me/projects/myapp/.ypk.pid

$ ypklog     # 实时看日志
INFO:     Started server process [12346]
INFO:     Uvicorn running on http://0.0.0.0:8080
INFO:     127.0.0.1:54321 - "GET / HTTP/1.1" 200 OK
^C          # Ctrl+C 退出 tail,服务不受影响

$ ypkstop
🛑 收到信号 15 (SIGTERM), 标记 user_stopped
🛑 停止 PID=12345...
✅ 已停止 PID=12345

高级用法

# 自定义模块(不在 main.py 里时)
ypkstart --module api.main:app

# 自定义端口/host(覆盖 .env)
ypkstart --port 9000 --host 127.0.0.1

# 开发模式(代码改动自动 reload)
ypkstart --reload

# 多环境配置(加载 .env.production + .env,后者被前者覆盖)
ypkstart --env production
# 也可用 APP_ENV 环境变量
APP_ENV=development ypkstart

# 关闭 supervisor 守护(回到旧行为,子进程被 OOM 杀不会自动重启)
ypkstart --no-supervisor

# 调内存阈值(默认 80% warn / 95% hard)
ypkstart --mem-warn-ratio 0.7 --mem-hard-ratio 0.9

# 调雪崩保护(默认 5 次 / 5 分钟)
ypkstart --max-restarts 10 --restart-window 600

# 端口被占时自动清理(杀端口上所有进程,危险!)
ypkstart --clean

# 用环境变量覆盖(适合 CI/CD)
YPK_MODULE=api:app PORT=9000 ypkstart

# 看更多帮助
ypk --help
ypk start --help

Supervisor 守护(为什么需要)

问题: Python 进程被 Linux OOM Killer 杀掉时,业务代码根本来不及反应 —— 是内核直接发 SIGKILL,无法被 Python 捕获。Java 的 -Xmx 这种"应用层堆限制"在 Python/CPython 里不存在,一旦进程 RSS 超过 OS 限制(物理机 / cgroup),进程直接消失。

解决: ypkstart 默认启一个 supervisor 进程守护 uvicorn:

调优建议:

# 容器/物理机部署(推荐配)
ypkstart \
  --mem-warn-ratio 0.7 \    # 70% 预警,留缓冲
  --mem-hard-ratio 0.85 \   # 85% 提前重启,给 OOM Killer 留余量
  --max-restarts 5 \         # 5 分钟内最多重启 5 次
  --restart-window 300       # 5 分钟窗口

关闭 supervisor:

# CI / 单次任务 / 想用 systemd / Docker 守护的场景
ypkstart --no-supervisor

推荐: 业务代码加 FastAPI lifespan 钩子处理 graceful shutdown,supervisor 发 SIGTERM 时能清理 DB 连接、排空请求:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动
    await db.connect()
    yield
    # 关闭时(supervisor 发 SIGTERM 触发)
    await db.disconnect()
    print("✅ 资源清理完成")

app = FastAPI(lifespan=lifespan)

行为约定

适用场景

• 本地开发: ypkstart --reload 一键启动,代码改动自动 reload
• 内网部署: nohup ypkstart & 或 ypkstart + disown,日志走 .ypk.log,状态 ypkstatus
• CI/CD: 环境变量覆盖 + ypkstop 在 job 结束时清理
• 每个项目不用写 start.sh / stop.sh: 统一用 ypkstart / ypkstop
• 生产环境: ypkstart --env production,supervisor 自动守护 OOM 重启

多环境配置(.env / .env.<env>)

支持一个项目多套环境配置,高优先级文件覆盖低优先级文件的同名变量。

⚠️ 重要: .env 是纯静态 KEY=VALUE 格式,不支持逻辑

# ✅ 合法:静态 KEY=VALUE
KEY=value
KEY="value with spaces"
export KEY=value
# 注释(行首 # 或 var=value 的 # 后面)

# ❌ 不支持:.env 里写 if / 变量插值 / 命令替换
if [ "$ENV" = "prod" ]; then ...    # 不会执行,被当成普通字符串
${OTHER_KEY}                        # 不会展开
$(somecommand)                      # 不会执行

所以多环境必须用多个 .env 文件,不能在单个 .env 里 if/else。

文件结构

your-project/
├── .env                  # 通用配置(所有环境共享,推荐放公共值)
├── .env.development      # 开发环境专属
├── .env.production       # 生产环境专属
└── .env.test             # 测试环境专属

加载优先级(--env production 时)

shell 环境变量 > .env.production > .env

同名变量高优先级覆盖低优先级:.env.production 里的 PORT 会覆盖 .env 里的 PORT。

示例

.env(公共,所有环境共享):

APP_NAME=MyApp
LOG_LEVEL=INFO
PORT=8000
# 其他环境无关的配置放这里

.env.production(只生产环境加载,覆盖 .env 的同名变量):

DB_HOST=prod-db.example.com
DB_PORT=5433
PORT=6001                       # 覆盖 .env 的 PORT=8000
APP_NAME=MyApp-production       # 覆盖 .env 的 APP_NAME

.env.development(只开发环境加载):

DB_HOST=localhost
DB_PORT=5432
LOG_LEVEL=DEBUG                 # 覆盖 .env 的 LOG_LEVEL=INFO

启动命令

# 启动开发环境(加载 .env.development + .env,前者覆盖后者)
ypkstart --env development
# 或
APP_ENV=development ypkstart

# 启动生产环境(加载 .env.production + .env)
ypkstart --env production
# 或
APP_ENV=production ypkstart

# 启动测试环境
APP_ENV=test ypkstart

# 默认(只加载 .env, 忽略所有 .env.<env>)
ypkstart

验证加载结果

启动时会打印当前加载了哪些文件:

$ APP_ENV=production ypkstart
📄 .env (env=production): ['.env.production', '.env'] (共新加载 12 个变量)
📦 模块: main:app
🌐 监听: 0.0.0.0:6001    ← 来自 .env.production,覆盖了 .env 的 8000
...

日志(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 友好

图片处理(yayo_pypkg.image_utils)

PDF 转图片工具,基于 PyMuPDF / fitz,可选依赖(在 [image] extras 里)。

.env 配置(无 —— 纯函数式,所有参数从 CLI / 函数入参传入):

# 装依赖
pip install yayo-pypkg[image]
# 或: uv add yayo-pypkg[image]
# 或: uv add pymupdf

基础用法(默认 PNG 200dpi,输出到 <pdf 同目录>/<pdf 名>_images/):

from yayo_pypkg.image_utils import pdf_to_images, pdf_page_count

# PDF → 一堆 PNG(每页一张)
images = pdf_to_images("report.pdf")
# → [Path("report_images/report_page_1.png"), ...]

# 拿总页数
n = pdf_page_count("report.pdf")

完整参数:

images = pdf_to_images(
    "report.pdf",
    output_dir="./outputs",     # 自定义输出目录
    dpi=150,                    # 自定义 DPI(默认 200)
    # zoom=2.0,                 # 也可以直接给缩放因子(优先级比 dpi 高)
    fmt="jpg",                  # png / jpg / jpeg(JPG 自动强制 alpha=False)
    first_page=1,               # 起始页(从 1 开始)
    last_page=3,                # 结束页(包含),只转 1-3 页
    alpha=False,                # 是否保留 alpha 通道(PNG 默认 False,JPG 自动 False)
)
# → [Path("./outputs/report_page_1.jpg"), Path("./outputs/report_page_2.jpg"), Path("./outputs/report_page_3.jpg")]

参数对照表:

特性:

• fitz 是可选依赖:没装时 import yayo_pypkg.image_utils 不报错,只有真的去调 pdf_to_images 才会给友好 ImportError
• 默认输出目录是 <pdf 同目录>/<pdf 名>_images/,避免污染源目录
• 进度日志:每转一页都打日志([N/M] page_X.png)
• PPTX → 图片:pptx_to_images() 通过系统 LibreOffice CLI 转(详见下方)

PPTX → 图片(pptx_to_images)

from yayo_pypkg.image_utils import pptx_to_images

images = pptx_to_images("slides.pptx")
# → [Path("slides_images/slides_slide_1.png"), ...]

images = pptx_to_images("slides.pptx", dpi=150, output_dir="./out")

系统依赖:需要装 LibreOffice(因为 Python 没原生 PPTX → 图片方案)

底层流程:

1. 调 soffice --headless --convert-to pdf 把 PPTX 转成临时 PDF
2. 用 :func:pdf_to_images 把临时 PDF 转成图片
3. 删临时 PDF(图片保留)

错误处理:

• 找不到 soffice 命令 → RuntimeError 提示装 LibreOffice
• LibreOffice 转 PDF 失败 → RuntimeError(含 stdout/stderr)
• 转 PDF 超时(默认 300s)→ RuntimeError
• 返回 list[Path](不是 str),用起来更方便
• 默认 capture_internal=True(内部 yayo_pypkg 日志也接到一起)

Redis(yayo_pypkg.db.db_redis)

同步 + 异步 双 API,命名跟 db_mysql.py 完全对齐 —— sync 方法无前缀,async 方法加 a 前缀。

API 选择 = 函数名(无配置项,无运行时切换):

• 想用 sync → 调 set_value(...) / init_redis()(任何 redis 版本都能用)
• 想用 async → 调 aset_value(...) / ainit_redis() 配合 await(必须 redis >= 4.2)

支持的 redis 版本:

.env 配置(sync 和 async 共用):

REDIS_ENABLE=true            # 开关
REDIS_HOST=127.0.0.1          # 单节点 host
REDIS_PORT=6379               # 单节点 port
REDIS_DB=0                    # 单节点 db
REDIS_PASSWORD=               # 密码(空 = 无密码)
REDIS_TIMEOUT=10              # 超时(秒)
REDIS_CLUSTER_NODES=          # 集群节点(非空自动走集群):host1:port,host2:port,host3:port

sync / async API 对照表

跟 db_mysql.py 的命名对照:init_db ↔ init_redis,ainit_db ↔ ainit_redis,get_db ↔ get_redis,aget_db ↔ aget_redis(以此类推)。

完整示例

.env(3 节点集群 + redis 3.5+ 内网):

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

业务代码(同步 + 异步混用,sync 永远能用,async 必须 redis 4.2+):

# ===== 1. sync API(脚本 / 定时任务 / FastAPI def 路由)=====
from yayo_pypkg.db.db_redis import (
    init_redis, close_redis, get_redis,
    set_value, get_value, set_json, get_json,
)

# 初始化
init_redis()

# 业务
set_value("foo", "bar")
print(get_value("foo"))               # 'bar'
set_json("user:1", {"name": "张三", "age": 18}, ttl=3600)
print(get_json("user:1"))             # {'name': '张三', 'age': 18}

# 清理
close_redis()

# ===== 2. async API(FastAPI async def 路由 / 异步任务,必须 redis 4.2+)=====
import asyncio
from yayo_pypkg.db.db_redis import (
    ainit_redis, aclose_redis, aget_redis,
    aset_value, aget_value, aset_json, aget_json,
)

async def main():
    await ainit_redis()
    try:
        await aset_value("foo", "bar")
        print(await aget_value("foo"))
        await aset_json("user:1", {"name": "张三"}, ttl=3600)
        print(await aget_json("user:1"))
    finally:
        await aclose_redis()

asyncio.run(main())

# ===== 3. FastAPI lifespan(必须 redis 4.2+,走 async)=====
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from yayo_pypkg.db.db_redis import ainit_redis, aclose_redis, aget_redis

@asynccontextmanager
async def lifespan(app):
    await ainit_redis()                   # redis 4.2+ 检查,3.x 直接 ImportError
    yield
    await aclose_redis()

app = FastAPI(lifespan=lifespan)

@app.get("/cache/{key}")
async def get_cache(key: str):
    r = aget_redis()                       # 拿 async 客户端
    val = await r.get(key)                 # 直接 await,走原生 redis.asyncio
    return {"value": val.decode() if val else None}

# ===== 4. FastAPI 同步 def 路由(任何 redis 版本,直接走 sync)=====
from fastapi import FastAPI
from yayo_pypkg.db.db_redis import init_redis, get_redis

app = FastAPI()

@app.on_event("startup")
def startup():
    init_redis()                           # 进程启动时 init 一次

@app.get("/cache-sync/{key}")
def cache_sync(key: str):
    r = get_redis()                        # 拿 sync 客户端
    return {"value": r.get(key).decode() if r.get(key) else None}

集群 vs 单节点(对业务代码透明)

集群注意事项:

• 多 key 操作(mget / delete / MSET 等)要保证所有 key 同一 hash slot,否则 CROSSSLOT
• 想强制一组 key 落到同 slot,用 hashtag:user:{1001}.name + user:{1001}.age → {} 里的 1001 是 hash tag

redis 版本怎么处理?

模块概览

MinerU OCR(yayo_pypkg.ocr.mineru_ocr)

异步 OCR 客户端,基于自建的 mineru-api。支持 PDF / 图片 单文件解析,新增加 parse_with_pages 支持带分页分隔符的 PDF / PPTX 文档解析。

.env 配置:

MINERU_ENABLE=true
MINERU_URL=http://127.0.0.1:8017
MINERU_TIMEOUT=600           # 可选,默认 600 秒
MINERU_BACKEND=hybrid-auto-engine  # 可选,默认 hybrid-auto-engine

单文件解析(MinerU.parse_file / MinerU.parse_image)

import asyncio
from yayo_pypkg.ocr.mineru_ocr import MinerU

client = MinerU()

async def main():
    # 文件(自动按扩展名派发)
    md = await client.parse_file("report.pdf")
    md = await client.parse_image("photo.jpg")
    
asyncio.run(main())

带分页分隔符的文档解析(MinerU.parse_with_pages)✨

MinerU 实例方法(跟 parse_file / parse_image 同级):

核心功能:把 PDF / PPTX 按页转图片,每页调 mineru OCR,然后用 XML 风格的分隔标签合并结果。

import asyncio
from yayo_pypkg.ocr.mineru_ocr import MinerU

client = MinerU()

async def main():
    # PDF → 每页用 <第 N 页>...</第 N 页> 包裹
    text = await client.parse_with_pages("report.pdf")
    print(text)
    # 输出示例:
    # <第 1 页>
    # 第一页 OCR 出来的 markdown 内容
    # </第 1 页>
    #
    # <第 2 页>
    # 第二页 OCR 出来的 markdown 内容
    # </第 2 页>

    # PPTX → 每张幻灯片用 <幻灯片 N>...</幻灯片 N> 包裹
    text = await client.parse_with_pages("slides.pptx")

    # 自定义 DPI / 保留图片 / 并发数
    text = await client.parse_with_pages(
        "report.pdf",
        dpi=150,
        keep_images=True,     # 保留中间生成的图片(默认 False 自动清理)
        concurrent=3,         # 并发调 mineru(默认 1 = 串行)
    )

asyncio.run(main())

完整流程(自动帮你跑):

PPT / PPTX file
    ↓ (image_utils.pdf_to_images / pptx_to_images)
[page_1.png, page_2.png, ...]
    ↓ (MinerU.parse_image × N)
["markdown_1", "markdown_2", ...]
    ↓ (合并 + 加 <第 N 页> 分隔符)
<第 1 页>\nmarkdown_1\n</第 1 页>\n\n<第 2 页>\n...

PPTX 降级行为 ⚠️:

PPTX 转图片需要系统装 LibreOffice(soffice 命令)。 内网很多机器可能没装,这时候:

• pptx_to_images() 抛 RuntimeError("找不到 soffice")
• parse_with_pages() 自动降级为直接调 self.parse_file() 整个 PPTX 当一个文件 OCR(不分页)
• 打 WARNING 日志说明原因和怎么修

[mineru] PPTX → 图片失败(LibreOffice 可能没装或转换失败): 系统 PATH 找不到 soffice / libreoffice 命令
          → 自动降级为不分页解析整个 PPTX(self.parse_file)。
          → 要恢复分页,请在系统装 LibreOffice:
             macOS: brew install --cask libreoffice
             Ubuntu: sudo apt-get install -y libreoffice

PDF 路径不会降级(因为 PDF 用 PyMuPDF,装 yayo-pypkg[image] 后保证能跑)。

系统依赖:

参数:

抛的异常:

• FileNotFoundError:文件不存在
• ValueError:格式不支持(只接受 PDF / PPTX)
• RuntimeError:缺 LibreOffice(PPTX)或 mineru API 失败

MySQL(yayo_pypkg.db.db_mysql)

同时提供同步 + 异步两套 API,业务代码按场景选。

.env 配置(两套共用):

DB_ENABLE=true
DB_HOST=127.0.0.1
DB_PORT=3306                   # 可选,默认 3306
DB_NAME=mydb
DB_USERNAME=root
DB_PASSWORD=secret
DB_POOL_SIZE=5                 # 可选,默认 5
DB_MAX_OVERFLOW=10             # 可选,默认 10
DB_POOL_RECYCLE=3600           # 可选,默认 3600(小于 MySQL 8h 超时)
DB_POOL_TIMEOUT=30             # 可选,默认 30

同步 API(给 def 路由 / 脚本 / 定时任务用)

from yayo_pypkg.db.db_mysql import init_db, get_db_session, get_db, close_db

# 启动时建表
init_db()                                  # 自动检测 SQLModel
init_db(models_dir="models")               # 显式扫描 models 目录
init_db(SQLModel)                          # 传统写法

# 业务代码(脚本 / 定时任务)
with get_db_session() as session:
    session.execute(text("SELECT 1"))

# FastAPI 同步路由
from fastapi import Depends
from yayo_pypkg.db.db_mysql import get_db

@app.get("/users")
def list_users(db = Depends(get_db)):
    return db.query(UserModel).all()

# 关闭
close_db()

异步 API(给 async def 路由用,不会阻塞 event loop)

from yayo_pypkg.db.db_mysql import ainit_db, aget_db_session, aget_db, aclose_db

# FastAPI lifespan
from contextlib import asynccontextmanager
from yayo_pypkg.db.db_mysql import ainit_db, aclose_db

@asynccontextmanager
async def lifespan(app):
    await ainit_db()                       # 异步建表
    yield
    await aclose_db()                      # 异步关闭连接池

# FastAPI 异步路由(SQLAlchemy 2.0 风格)
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from yayo_pypkg.db.db_mysql import aget_db

@app.get("/users")
async def list_users(db: AsyncSession = Depends(aget_db)):
    # 不会阻塞 event loop ✅
    result = await db.execute(select(UserModel))
    return result.scalars().all()

# 异步上下文管理器(async def 业务函数)
async def my_business_logic():
    async with aget_db_session() as session:
        result = await session.execute(select(UserModel))
        users = result.scalars().all()
    # 自动 commit

sync vs async 选哪个?

反模式(要避免):async def 路由里 Depends(get_db)(同步) → 阻塞整个 event loop。

License

MIT