pyre-sqlalchemy-db

基于 SQLAlchemy 2.0 的轻量 ORM 增强库：声明式字段、事务管理、分页、软删除、雪花 ID 以及完整的异步支持，并与 Flask / Starlette 等框架无侵入集成。

✨ 特点

• 🧱 声明式字段 — StringField、DateTimeField、ForeignKeyField 开箱即用，参数透传 mapped_column。
• 🔁 事务管理 — transactional / local_transaction，支持 REQUIRED / NEW / MANDATORY 传播与隔离级别。
• ⚡ 异步支持 — async_transactional / Model.async_query，同一套模型同步、异步通用。
• 🧩 常用 Mixin — 主键、创建/更新时间、软删除、操作人，按需组合。
• 🆔 主键策略 — 自增、雪花 ID、UUID 三种，一行切换。
• 🔌 多数据源 — 按 schema 绑定多个引擎，模型自动路由。
• 📤 序列化 — to_dict() 内置循环引用保护与日期 / Decimal / 枚举编码。
• 🪶 框架无关 — 核心不依赖任何 Web 框架，可选接入 Flask 应用上下文。

📦 安装

pip install pyre-sqlalchemy-db

按需安装可选依赖：

pip install "pyre-sqlalchemy-db[async]"   # greenlet / aiosqlite / aiomysql / asyncpg
pip install "pyre-sqlalchemy-db[mysql]"
pip install "pyre-sqlalchemy-db[flask]"

💡 一个简单的例子

from sqlalchemy_db import SQLAlchemyDB, models

db = SQLAlchemyDB("sqlite:///demo.db")


class User(models.PkMixin, models.TimestampMixin, models.Model):
    __tablename__ = "user"

    name = models.StringField(50, comment="用户名")
    age = models.IntegerField(default=0)

字段、模型与 Mixin 统一从 models 命名空间取用（models 即 sqlalchemy_db.models 子包）。

👩‍💻 目录结构

sqlalchemy-db
├── sqlalchemy_db
│   ├── __init__.py           # 对外导出
│   ├── app.py                # 框架集成入口
│   ├── transaction.py        # 同步事务
│   ├── async_transaction.py  # 异步事务
│   ├── snowflake.py          # 雪花 ID 生成
│   ├── encoders.py           # 序列化编码器
│   ├── exceptions.py         # 异常体系
│   ├── models/               # Model / 字段 / Mixin / 查询
│   ├── globals.py            # 实例解析
│   ├── constants.py
│   └── utils.py
├── examples/                 # 可运行示例
├── tests/                    # 单元测试
└── setup.py

🚀 快速开始

from sqlalchemy_db import SQLAlchemyDB, Page, local_transaction

db = SQLAlchemyDB("sqlite:///demo.db")
db.create_all()

# 写入
with local_transaction():
    user = User(name="Alice", age=18)
    user.insert()
    print(user.id)          # 1

# 更新与删除
with local_transaction():
    user.save()
    user.delete()

# 查询与分页
with local_transaction():
    User.query.get_by_id_or_404(1)
    User.query.select(User.age >= 18, order_by=User.id.desc())

    page = Page(page=1, page_size=20)
    User.query.select_page(page)
    page.get_data()             # 数据列表
    page.get_count()            # 总条数

Model.query 在访问时即从当前会话上下文取 Session，因此所有查询都要运行在 local_transaction() / transactional 上下文内。否则库会告警事务上下文不会自动回收。

🧱 模型

字段

所有字段均透传 mapped_column 参数，nullable、comment、server_default、index、 unique、sort_order 照常可用。

Mixin

主键策略

from sqlalchemy_db import PkType, models


class Order(models.Model):
    __tablename__ = "order"
    id = models.AutoPkField(pk_type=PkType.SNOWFLAKE)

序列化

user.to_dict()
user.to_dict(exclude={"age"}, exclude_none=True)

User.from_dict({"name": "Bob", "age": 20})
User.from_entity(pydantic_model)

🔍 查询

Model.query / Model.async_query 提供统一的链式查询入口。查询须在会话上下文内执行：

with local_transaction():
    User.query.get_by_id(1)                       # 按主键
    User.query.get_by_id_or_404(1)                # 不存在抛 NotFoundDataException
    User.query.get_one(User.name == "Alice")
    User.query.select(User.age >= 18, order_by=User.id.desc())
    User.query.select_batch_ids([1, 2, 3])
    User.query.select_count(User.is_active.is_(True))
    User.query.delete_by_id_or_404(1)

软删除：

with local_transaction():
    User.query.with_not_trashed().all()   # 排除已删除
    User.query.only_trashed().all()       # 仅已删除
    User.query.with_trashed().all()       # 全部

🔁 事务

from sqlalchemy_db import transactional, local_transaction, Propagation, Isolation


@transactional
def create_user(name: str):
    User(name=name).insert()


with local_transaction():
    User(name="Bob").insert()


@transactional(propagation=Propagation.NEW, isolation=Isolation.READ_COMMITTED, autocommit=True)
def task():
    ...

⚡ 异步

模型无需改动，异步场景使用 async_transactional 与 Model.async_query：

from sqlalchemy_db import SQLAlchemyDB, async_transactional

db = SQLAlchemyDB("sqlite+aiosqlite:///demo.db")


@async_transactional
async def main():
    user = await User.async_query.get_by_id(1)
    rows = await User.async_query.select(User.age >= 18)
    page = await User.async_query.select_page(Page(page=1, page_size=20))

⚙️ 配置

db = SQLAlchemyDB({
    "SQLALCHEMY_DATABASE_URI": "mysql+pymysql://root:123456@127.0.0.1:3306/demo",
    "SQLALCHEMY_ECHO": False,
    "SQLALCHEMY_ENGINE_OPTIONS": {"pool_size": 10, "pool_recycle": 3600},
    "SQLALCHEMY_ASYNC_MODE": "auto",          # auto / sync / async
    "SNOWFLAKE_DATACENTER_ID": 1,
    "SNOWFLAKE_WORKER_ID": 1,
})

🔌 多数据源

from sqlalchemy_db import models


class Log(models.Model):
    __tablename__ = "log"
    __schema__ = "log"        # 路由到 "log" 数据源

db = SQLAlchemyDB({
    "SQLALCHEMY_DATABASE_URI": "mysql+pymysql://.../main",
    "SQLALCHEMY_SCHEMAS": {"log": "mysql+pymysql://.../logs"},
})

🧩 Flask 集成

from flask import Flask
from sqlalchemy_db import SQLAlchemyDB

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:123456@127.0.0.1:3306/demo"

db = SQLAlchemyDB(app)


@app.route("/users/<int:uid>")
def get_user(uid):
    return db.session.get(User, uid).to_dict()

位于 Flask 应用上下文内时，Model.query 等入口会自动解析当前应用绑定的实例。

📚 示例

更多可运行示例见 examples/。

📄 许可

本项目基于 MIT 许可开源。