yayo-pypkg 0.0.12

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]

# 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]

快速上手

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

# 方式 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 自动读取(支持类型自动转换)

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

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

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

ypkstart                          # 启动服务(后台)
ypkstatus                         # 查看状态
ypkstop                           # 停止服务
ypklog                            # tail -f 日志(Ctrl+C 退出)
ypk start --reload                # 开发模式(代码改动自动 reload)

自动做的事

ypkstart 会自动:

1. 在 CWD 找 .env,找不到在上级目录找(适合 monorepo)
2. 解析 .env 加载到 os.environ(已有变量不覆盖,shell 环境变量优先)
3. 找入口模块:main.py / app.py / server.py(按顺序)
4. 从 .env 读 PORT 或 YPK_PORT(host 同理)
5. 后台启动 uvicorn,日志写 .ypk.log,PID 写 .ypk.pid
6. 等 0.5 秒探活,启动失败就清理 PID 文件

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

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

完整示例

$ cd ~/projects/myapp
$ ypkstart
📄 .env: /home/me/projects/myapp/.env (新加载 5 个变量)
📦 模块: main:app
🌐 监听: 0.0.0.0:8080
🚀 启动命令: /path/to/python -m uvicorn main:app --host 0.0.0.0 --port 8080
✅ 启动成功!
   PID:  12345
   端口: 8080
   日志: /home/me/projects/myapp/.ypk.log
   PID 文件: /home/me/projects/myapp/.ypk.pid

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

$ ypkstatus
✅ 正在跑 PID=12345
   端口(从 .env): 8080
   日志: /home/me/projects/myapp/.ypk.log

$ ypklog     # 实时看日志
INFO:     Started server process [12345]
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
🛑 停止 PID=12345...
✅ 已停止 PID=12345

高级用法

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

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

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

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

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

行为约定

配置	默认行为	覆盖方式
入口模块	CWD 下找 main.py / app.py / server.py	--module api:app 或 YPK_MODULE=api:app
端口	.env 的 PORT 或 YPK_PORT → 默认 8000	--port 9000
主机	.env 的 HOST 或 YPK_HOST → 默认 0.0.0.0	--host 127.0.0.1
.env 路径	<cwd>/.env → <parent>/.env	不支持(以后可加 --env)
PID 文件	<cwd>/.ypk.pid	不可改
日志文件	<cwd>/.ypk.log	不可改

适用场景

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

日志(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")]

参数对照表:

参数	默认	说明
pdf_path	(必填)	PDF 文件路径(str / Path 都行)
output_dir	<pdf 同目录>/<pdf 名>_images/	输出目录
dpi	200	图片分辨率(PDF 标准是 72 dpi)
zoom	None	缩放因子(直接给 fitz.Matrix)。zoom 和 dpi 同时给时,zoom 优先
fmt	"png"	"png" / "jpg" / "jpeg"
first_page	1	起始页(从 1 开始)
last_page	None	结束页,None 表示到最后
alpha	False	是否保留 alpha 通道(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 → 图片方案)

OS	安装命令
macOS	brew install --cask libreoffice
Ubuntu/Debian	sudo apt-get install -y libreoffice
Windows	从 libreoffice.org 下载安装

底层流程:

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 版本:

redis 版本	sync API	async API	适用环境
redis 3.5+	✅	❌ ImportError	老内网 / JupyterHub / 老 Conda 环境(你的内网 redis 3.5.3 就是这种)
redis 4.2+	✅	✅	新项目 / 生产
redis 5.0+(Python 3.7+ 推荐)	✅	✅	推荐
redis 6.0+(Python 3.8+)	✅	✅	最新

.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 对照表

操作	sync(任何 redis 版本)	async(必须 redis 4.2+)
初始化 / 关闭	init_redis() / close_redis() / get_redis()	ainit_redis() / aclose_redis() / aget_redis()
写 KV	set_value(key, value, ttl=None)	await aset_value(...)
读 KV	get_value(key)	await aget_value(...)
删 key	delete(*keys)	await adelete(...)
查存在	exists(*keys)	await aexists(...)
设过期	expire(key, ttl)	await aexpire(...)
批量读	mget(*keys)	await amget(...)
写 JSON	set_json(key, obj, ttl=None)	await aset_json(...)
读 JSON	get_json(key)	await aget_json(...)
健康检查	ping()	await aping()

跟 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 单节点(对业务代码透明)

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

集群注意事项:

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

redis 版本怎么处理?

你的情况	怎么搞
内网 redis 3.5.3(老环境)	装 redis>=3.5 → 用 sync API(任何版本都行)
想用 async 性能最优	pip install 'redis>=4.2' → sync + async 双 API
生产新项目	pip install 'redis>=5.0'(Python 3.7+ 最稳)
别人代码用错了 async 在 3.x	报错会清楚告诉你"当前是 X.X,需要 4.2+,改用 sync API"

模块概览

模块	说明
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.image_utils	PDF 转图片([image] extras,基于 PyMuPDF)
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(单节点 + 集群,[api] extras)
yayo_pypkg.ocr	MinerU OCR 客户端

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] 后保证能跑)。

系统依赖:

格式	依赖
PDF	pymupdf(pip install yayo-pypkg[image])+ mineru-api
PPTX	pymupdf + 系统装 LibreOffice(brew install --cask libreoffice)+ mineru-api
图片	mineru-api

参数:

参数	默认	说明
file_path	(必填)	PDF / PPTX 路径
image_output_dir	None	中间图片输出目录,默认在临时目录,跑完自动删
dpi	200	转图片的分辨率
keep_images	False	是否保留中间图片
concurrent	1	并发调 mineru 的图片数(>1 提速但小心 API 限流)
timeout	300	LibreOffice 转 PDF 超时(秒)

抛的异常:

• 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 选哪个?

业务路由类型	用	为什么
def 路由(无 await)	同步 get_db / get_db_session	FastAPI 自动丢线程池,def 路由调同步 DB 不会阻塞 event loop
async def 路由(有 await)	异步 aget_db / aget_db_session	同步 DB 会阻塞 event loop,async 路由必须配 async DB
脚本 / 定时任务	同步	没 event loop 的概念,同步更简单

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

License

MIT