ORM

大家都知道ORM（Object Relational Mapping）是一种将对象和关系数据库中的表进行映射的技术，它可以让开发者更加方便地操作数据库，而不用直接使用SQL语句。

直接使用SQL语句操作数据库，虽然可以让开发者直接与数据库打交道，但手动编写SQL语句，容易出错，而且灵活性上比较欠缺。相比之下，使用ORM（以SQLAlchemy为例）有更加易于使用、更加灵活、能防止 SQL 注入攻击、更加易于测试的优势。

当然，使用 SQLAlchemy 也会增加代码的复杂度，需要学习额外的知识和 API。因此，在实际应用中需要根据具体情况进行选择。

那么有没有一种技术或者框架 既不用增加太多的应用成本，又兼具以SQLAlchemy为代表的ORM 框架的优势 呢？答案是肯定的，那就是我们今天介绍的主角 sqlmodel.

我们就以 Fastapi 开发创建用户和查询用户 两个功能的接口来对比一下 ，SQLAlchemy 和 sqlmodel， sqlmodel 和 只使用 SQL的差异。

使用SQLAlchemy

安装

pip install sqlalchemy

示例代码

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import Session, declarative_base, sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql://user:password@host:port/database"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

app = FastAPI()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    age = Column(Integer)

class UserIn(Base):
    name: str
    age: int

class UserOut(Base):
    id: int
    name: str
    age: int

class UserUpdate(Base):
    name: Optional[str] = None
    age: Optional[int] = None

Base.metadata.create_all(bind=engine)

def get_db():
    db = None
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()

def create_user(db: Session, user: UserIn):
    db_user = User(name=user.name, age=user.age)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def read_user(db: Session, user_id: int):
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

def read_all_user(db: Session, ):
    db_user = db.query(User).all()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.post("/users/", response_model=UserOut)
async def create_user_view(user: UserIn, db: Session = Depends(get_db)):
    return create_user(db, user)

@app.get("/users/{user_id}", response_model=UserOut)
async def read_user_view(user_id: int, db: Session = Depends(get_db)):
    return read_user(db, user_id)

@app.get("/users/", response_model=UserOut)
async def read_all_user_view(db: Session = Depends(get_db)):
    return read_all_user(db)

代码解释

User 是数据模型类的名称，id、name、age 是表中的列名。UserIn 是创建用户的请求参数模型，UserOut 是查询用户的响应数据模型，UserUpdate 是更新用户的请求参数模型。

使用 create_engine 函数创建一个数据库连接引擎，使用 sessionmaker 函数创建一个数据库会话工厂，使用 declarative_base 函数创建一个基类。在创建表时，使用 Base.metadata.create_all 函数创建表。

使用 get_db 函数获取数据库会话对象，使用 create_user 和 read_user 函数进行数据库操作。在视图函数中，只需要调用这些函数即可完成相应的业务逻辑。

上面的代码已经非常简洁直观，但是还是有有一定的学习成本，下面我们来看下使用我们今天的主角 -- sqlmodel 需要怎样来实现上面的接口。

使用sqlmodel

安装 sqlmodel

pip install sqlmodel

示例代码

# -*- coding: utf-8 -*-
"""
@File   :dda.py
@Date   :2023-06-05
@user   :bingoHe
"""
from typing import Optional

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlmodel import SQLModel, Field, create_all, Session as SQLModelSession

SQLALCHEMY_DATABASE_URL = "mysql://user:password@host:port/database"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

app = FastAPI()

class UserBase(SQLModel):
    name: Optional[str] = None
    age: Optional[int] = None

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class UserIn(UserBase):
    pass

class UserOut(UserBase):
    id: int

class UserUpdate(UserBase):
    pass


create_all(engine)

def get_db():
    """获取数据库会话对象"""
    db = None
    try:
        db = SQLModelSession(engine)
        yield db
    finally:
        db.close()

def create_user(db: SQLModelSession, user: UserIn):
    """创建用户"""
    db_user = User.from_orm(user)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def read_user(db: SQLModelSession, user_id: int):
    """查询用户"""
    db_user = db.get(User, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn, db: SQLModelSession = Depends(get_db)):
    """创建用户"""
    return create_user(db, user)

@app.get("/users/{user_id}", response_model=UserOut)
async def read_user(user_id: int, db: SQLModelSession = Depends(get_db)):
    """查询用户"""
    return read_user(db, user_id)

和SQLAlchemy的主要使用差异在参数的定义上，使用多处继承，而不是各自定义的方法：

# Code above omitted 👆
...
class UserBase(SQLModel):
    name: Optional[str] = None
    age: Optional[int] = None

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class UserIn(UserBase):
    pass

class UserOut(UserBase):
    id: int

class UserUpdate(UserBase):
    pass
...
# Code below omitted 👇

继承这一点对于还在频繁迭代的系统中非常重要，因为同样添加 一个user的数据结构，SQLAlchemy需要修改4处地方，而sqlmodel 仅仅只需要修改一处。如果有多个表，这个便利性的优势会尤为突出。

这也就引出了sqlmodel具有的优势：

• 简短： 最小化代码重复。一个单一的类型注解做了很多工作。无需在 SQLAlchemy 和 Pydantic 中复制模型。
• 简单易用： API 设计简单易用，强大的编辑器支持，学习曲线较低，可以快速上手。它使用 Python 类型注解来定义数据模型，可以自动推断数据库表结构，同时支持类型检查和数据验证。
• 可扩展： 拥有 SQLAlchemy 和 Pydantic 的所有功能。
• 高性能： sqlmodel 采用了一些性能优化策略，比如使用预编译 SQL 语句、减少数据库连接次数等，可以提高数据库操作的性能。
• 支持异步操作： sqlmodel 支持异步操作，可以与 asyncio 库一起使用，可以在高并发场景下提高程序的性能。
• 支持原生 SQL： sqlmodel 支持原生 SQL，可以使用原生 SQL 语句进行数据库操作，同时还支持参数绑定和 SQL 注入防护。

SQLModel 实际上是在 Pydantic 和 SQLAlchemy 之间增加了一层兼容适配，经过精心设计以兼容两者。SQLModel 旨在简化 FastAPI 应用程序中与 SQL 数据库的交互。它结合了 SQLAlchemy 和 Pydantic，并尝试尽可能简化代码，让代码重复减少到最低限度，同时尽可能让开发人员获得最佳的开发体验。

原生的SQL语句支持举例

有时候我们可能需要使用原生的SQL语句来进行一些复杂的操作。

from sqlmodel import create_engine, Session

# 创建数据库引擎
engine = create_engine("sqlite:///example.db")

# 创建Session对象
with Session(engine) as session:
    # 执行原生的SQL语句
    result = session.execute("SELECT * FROM users WHERE age > :age", {"age": 18})

    # 处理查询结果
    for row in result:
        print(row)

高级用法：结合mixin类，简化数据库操作

结合mixin类，简化数据库操作，一处封装，处处适用。

如果熟悉fastapi，且仔细观察上面的完整代码就会发现，除了下面这段，其他的都是标准的Fastapi 接口开发需要的信息。而这样的操作结合我们接下来介绍的mixin方法，就可以给这只虎添上翅膀。

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

tips: 在面向对象编程中，Mixin是一种重用代码的方式，它是一个类，包含一些方法和属性，可以被其他类继承和使用。Mixin类通常不是独立的类，而是用于增强其他类的功能。Mixin类的优点在于可以将代码分解为小的、可重用的部分，从而减少代码的重复和冗余。Mixin类可以被多个类继承，从而避免了多重继承的问题。

import uvicorn
from typing import Optional, Union

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Field, Session, SQLModel, create_engine, select


class ActiveRecord(SQLModel):
    @classmethod
    def by_id(cls, _id: int, session):
        obj = session.get(cls, _id)
        if obj is None:
            raise HTTPException(status_code=404, detail=f"{cls.__name__} with id {id} not found")
        return obj

    @classmethod
    def all(cls, session):
        return session.exec(select(cls)).all()

    @classmethod
    def create(cls, source: Union[dict, SQLModel], session):
        if isinstance(source, SQLModel):
            obj = cls.from_orm(source)
        # elif isinstance(source, dict):
        elif isinstance(source, dict):
            obj = cls.parse_obj(source)
        session.add(obj)
        session.commit()
        session.refresh(obj)
        return obj

    def save(self, session):
        session.add(self)
        session.commit()
        session.refresh(self)


class UserBase(SQLModel):
    name: Optional[str] = None
    age: Optional[int] = None


class User(UserBase, ActiveRecord, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    __table_args__ = {'extend_existing': True}


class UserIn(UserBase):
    pass


class UserOut(UserBase):
    id: int


class UserUpdate(UserBase):
    pass

# 注意：需要提前安装pymysql， pip install pymysql
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://user:password@host:port/database"

engine = create_engine(SQLALCHEMY_DATABASE_URL)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/User/", response_model=UserOut)
def create_user(hero: UserIn, session: Session = Depends(get_session)):
    return User.create(hero, session)

@app.get("/User/", response_model=list[UserOut])
def read_user(session: Session = Depends(get_session)):
    return User.all(session)

@app.get("/User/{user_id}", response_model=UserOut)
def read_user(user_id: int,session: Session = Depends(get_session)):
    return User.by_id(user_id, session)


if __name__ == '__main__':
    uvicorn.run("main:app", reload=True)

总结

使用SQLModel + mixins可以在公共的逻辑里面实现增删改查操作，处封装，处处适用，减少了代码的重复性和冗余性。

建议：

根据上述比较，我们可以得出以下选择建议：

• 如果需要使用全面的ORM功能，或者需要使用复杂的查询构建和事务处理等功能，建议选择SQLAlchemy。
• 如果需要使用轻量级的ORM功能，或者需要快速上手和使用，建议选择sqlmodel。
• 如果需要支持多种数据库，建议两者都可以考虑使用。
• 如果对文档和社区支持有较高的要求，建议选择SQLAlchemy。
• 如果对代码规范有较高的要求，建议选择sqlmodel。