3

【Python】sqlmodel: Python 数据库管理ORM 的终极形态？ - _BingoHe

 1 year ago
source link: https://www.cnblogs.com/Detector/p/17458890.html
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

阅读目录

欢迎转载，也请在明显位置注明出处 ：https://www.cnblogs.com/Detector/p/17458890.html 谢谢！

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

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

点击查看优势说明

更加易于使用: 可以使用 Python 对象来表示数据库中的表和行，而不是直接使用 SQL 语句。这样可以使代码更加易于编写和维护。
更加灵活: SQLAlchemy 提供了灵活的查询语言，可以通过链式调用的方式构建复杂的查询语句。同时，SQLAlchemy 支持多种数据库，可以在不同的数据库之间进行切换，而不需要修改代码。
防止 SQL 注入攻击: SQLAlchemy 提供了参数化查询的方式，可以有效地防止 SQL 注入攻击。使用参数化查询可以将用户输入的数据转换为参数，从而避免了 SQL 注入攻击。
更加易于测试: 使用 SQLAlchemy 可以将业务逻辑和数据库操作分离，从而使得代码更加易于测试。可以通过 Mock 对象模拟数据库操作，从而进行单元测试和集成测试。
...

当然，使用 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可以在公共的逻辑里面实现增删改查操作，处封装，处处适用，减少了代码的重复性和冗余性。

特点 SQLAlchemy sqlmodel
数据库支持 支持多种数据库，包括MySQL、PostgreSQL、SQLite等 支持多种数据库，包括MySQL、PostgreSQL、SQLite等
ORM功能 提供全面的ORM功能，支持对象关系映射、事务处理、查询构建等 提供轻量级的ORM功能，支持对象关系映射、查询构建等
性能 性能较好，支持缓存、连接池等优化手段 性能较好，支持缓存、连接池等优化手段
学习难度 学习曲线较陡峭，需要掌握复杂的概念和API 学习曲线较平缓，易于上手和使用
文档和社区支持 提供完善的文档和活跃的社区支持 文档和社区支持相对较少
代码规范 代码规范较为灵活，可以自由组织代码结构 代码规范较为严格，需要按照规范组织代码结构

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

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

report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK