【Python】sqlmodel: Python 数据库管理ORM 的终极形态？
<h2 id="orm">ORM</h2><p>大家都知道ORM（Object Relational Mapping）是一种将对象和关系数据库中的表进行映射的技术，它可以让开发者更加方便地操作数据库，而不用直接使用SQL语句。</p>
<p>直接使用SQL语句操作数据库，虽然可以让开发者直接与数据库打交道，但手动编写SQL语句，容易出错，而且灵活性上比较欠缺。相比之下，使用ORM（以<code>SQLAlchemy</code>为例）有更加易于使用、更加灵活、能防止 <code>SQL</code> 注入攻击、更加易于测试的优势。</p>
<details>
<summary><strong style="color: rgba(0, 0, 255, 1)">点击查看优势说明 </strong></summary>
<blockquote>
<p><strong>更加易于使用:</strong> 可以使用 Python 对象来表示数据库中的表和行，而不是直接使用 SQL 语句。这样可以使代码更加易于编写和维护。<br>
<strong>更加灵活:</strong> SQLAlchemy 提供了灵活的查询语言，可以通过链式调用的方式构建复杂的查询语句。同时，SQLAlchemy 支持多种数据库，可以在不同的数据库之间进行切换，而不需要修改代码。<br>
<strong>防止 SQL 注入攻击:</strong> SQLAlchemy 提供了参数化查询的方式，可以有效地防止 SQL 注入攻击。使用参数化查询可以将用户输入的数据转换为参数，从而避免了 SQL 注入攻击。<br>
<strong>更加易于测试:</strong> 使用 SQLAlchemy 可以将业务逻辑和数据库操作分离，从而使得代码更加易于测试。可以通过 Mock 对象模拟数据库操作，从而进行单元测试和集成测试。<br>
...</p>
</blockquote>
</details>
<p>当然，使用 <code>SQLAlchemy</code> 也会增加代码的复杂度，需要学习额外的知识和 API。因此，在实际应用中需要根据具体情况进行选择。</p>
<p>那么有没有一种技术或者框架 <strong>既不用增加太多的应用成本，又兼具以<code>SQLAlchemy</code>为代表的ORM 框架的优势</strong> 呢？答案是肯定的，那就是我们今天介绍的主角 <strong><code>sqlmodel</code></strong>.</p>
<p>我们就以 <code>Fastapi</code> 开发<strong>创建用户</strong>和<strong>查询用户</strong> 两个功能的接口来对比一下 ，<code>SQLAlchemy</code> 和 <code>sqlmodel</code>， <code>sqlmodel</code> 和 只使用 <code>SQL</code>的差异。</p>
<h2 id="使用sqlalchemy">使用SQLAlchemy</h2>
<h4 id="安装">安装</h4>
<pre><code>pip install sqlalchemy
</code></pre>
<h4 id="示例代码">示例代码</h4>
<pre><code class="language-python">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 = None
age: Optional = 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)
</code></pre>
<h4 id="代码解释">代码解释</h4>
<p><code>User</code> 是数据模型类的名称，<code>id</code>、<code>name</code>、<code>age</code> 是表中的列名。<code>UserIn</code> 是创建用户的请求参数模型，<code>UserOut</code> 是查询用户的响应数据模型，<code>UserUpdate</code> 是更新用户的请求参数模型。</p>
<p>使用 <code>create_engine</code> 函数创建一个数据库连接引擎，使用 <code>sessionmaker</code> 函数创建一个数据库会话工厂，使用 <code>declarative_base</code> 函数创建一个基类。在创建表时，使用 <code>Base.metadata.create_all</code> 函数创建表。</p>
<p>使用 <code>get_db</code> 函数获取数据库会话对象，使用 <code>create_user</code> 和 <code>read_user</code> 函数进行数据库操作。在视图函数中，只需要调用这些函数即可完成相应的业务逻辑。</p>
<p>上面的代码已经非常简洁直观，但是还是有有一定的学习成本，下面我们来看下使用我们今天的主角 -- <code>sqlmodel</code> 需要怎样来实现上面的接口。</p>
<h2 id="使用sqlmodel">使用sqlmodel</h2>
<h4 id="安装-sqlmodel">安装 sqlmodel</h4>
<pre><code>pip install sqlmodel
</code></pre>
<h4 id="示例代码-1">示例代码</h4>
<hr>
<details>
<summary><strong style="color: rgba(219, 112, 147, 1)">点击查看完整代码</strong></summary>
<pre><code class="language-python"># -*- 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 = None
age: Optional = None
class User(UserBase, table=True):
id: Optional = 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)
</code></pre>
</details>
<hr>
<p>和<code>SQLAlchemy</code>的主要使用差异在参数的定义上，使用多处继承，而不是各自定义的方法：</p>
<pre><code class="language-python"># Code above omitted 👆
...
class UserBase(SQLModel):
name: Optional = None
age: Optional = None
class User(UserBase, table=True):
id: Optional = Field(default=None, primary_key=True)
class UserIn(UserBase):
pass
class UserOut(UserBase):
id: int
class UserUpdate(UserBase):
pass
...
# Code below omitted 👇
</code></pre>
<p>继承这一点对于还在频繁迭代的系统中非常重要，因为同样添加 一个user的数据结构，SQLAlchemy需要修改4处地方，而<code>sqlmodel</code> 仅仅只需要修改一处。如果有多个表，这个便利性的优势会尤为突出。</p>
<p>这也就引出了<code>sqlmodel</code>具有的优势：</p>
<ul>
<li><strong>简短：</strong>最小化代码重复。一个单一的类型注解做了很多工作。无需在 SQLAlchemy 和 <code>Pydantic</code> 中复制模型。</li>
<li><strong>简单易用：</strong>API 设计简单易用，强大的编辑器支持，学习曲线较低，可以快速上手。它使用 Python 类型注解来定义数据模型，可以自动推断数据库表结构，同时支持类型检查和数据验证。</li>
<li><strong>可扩展：</strong>拥有 <code>SQLAlchemy</code> 和 <code>Pydantic</code> 的所有功能。</li>
<li><strong>高性能</strong>： <code>sqlmodel</code> 采用了一些性能优化策略，比如使用预编译 SQL 语句、减少数据库连接次数等，可以提高数据库操作的性能。</li>
<li><strong>支持异步操作：</strong><code>sqlmodel</code> 支持异步操作，可以与 asyncio 库一起使用，可以在高并发场景下提高程序的性能。</li>
<li><strong>支持原生 SQL：</strong><code>sqlmodel</code> 支持原生 SQL，可以使用原生 SQL 语句进行数据库操作，同时还支持参数绑定和 SQL 注入防护。</li>
</ul>
<blockquote>
<p>SQLModel 实际上是在 Pydantic 和 SQLAlchemy 之间增加了一层兼容适配，经过精心设计以兼容两者。SQLModel 旨在简化 FastAPI 应用程序中与 SQL 数据库的交互。它结合了 SQLAlchemy 和 Pydantic，并尝试尽可能简化代码，让代码重复减少到最低限度，同时尽可能让开发人员获得最佳的开发体验。</p>
</blockquote>
<h4 id="原生的sql语句支持举例">原生的SQL语句支持举例</h4>
<p>有时候我们可能需要使用原生的SQL语句来进行一些复杂的操作。</p>
<pre><code class="language-python">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)
</code></pre>
<h2 id="高级用法结合mixin类简化数据库操作">高级用法：结合mixin类，简化数据库操作</h2>
<p>结合mixin类，简化数据库操作，一处封装，处处适用。</p>
<h5 id="如果熟悉fastapi且仔细观察上面的完整代码就会发现除了下面这段其他的都是标准的fastapi-接口开发需要的信息而这样的操作结合我们接下来介绍的mixin方法就可以给这只虎添上翅膀"><strong>如果熟悉fastapi，且仔细观察上面的完整代码就会发现，除了下面这段，其他的都是标准的<code>Fastapi</code> 接口开发需要的信息。而这样的操作结合我们接下来介绍的<code>mixin</code>方法，就可以给这只虎添上翅膀。</strong></h5>
<pre><code class="language-python">class User(UserBase, table=True):
id: Optional = Field(default=None, primary_key=True)
</code></pre>
<blockquote>
<p><strong>tips</strong>: 在面向对象编程中，Mixin是一种重用代码的方式，它是一个类，包含一些方法和属性，可以被其他类继承和使用。Mixin类通常不是独立的类，而是用于增强其他类的功能。Mixin类的优点在于可以将代码分解为小的、可重用的部分，从而减少代码的重复和冗余。Mixin类可以被多个类继承，从而避免了多重继承的问题。</p>
</blockquote>
<pre><code class="language-python">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, 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 = None
age: Optional = None
class User(UserBase, ActiveRecord, table=True):
id: Optional = 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)
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)
</code></pre>
<h2 id="总结">总结</h2>
<p>使用<code>SQLModel</code> + <code>mixins</code>可以在公共的逻辑里面实现增删改查操作，处封装，处处适用，减少了代码的重复性和冗余性。</p>
<table>
<thead>
<tr>
<th>特点</th>
<th>SQLAlchemy</th>
<th>sqlmodel</th>
</tr>
</thead>
<tbody>
<tr>
<td>数据库支持</td>
<td>支持多种数据库，包括MySQL、PostgreSQL、SQLite等</td>
<td>支持多种数据库，包括MySQL、PostgreSQL、SQLite等</td>
</tr>
<tr>
<td>ORM功能</td>
<td>提供全面的ORM功能，支持对象关系映射、事务处理、查询构建等</td>
<td>提供轻量级的ORM功能，支持对象关系映射、查询构建等</td>
</tr>
<tr>
<td>性能</td>
<td>性能较好，支持缓存、连接池等优化手段</td>
<td>性能较好，支持缓存、连接池等优化手段</td>
</tr>
<tr>
<td>学习难度</td>
<td>学习曲线较陡峭，需要掌握复杂的概念和API</td>
<td>学习曲线较平缓，易于上手和使用</td>
</tr>
<tr>
<td>文档和社区支持</td>
<td>提供完善的文档和活跃的社区支持</td>
<td>文档和社区支持相对较少</td>
</tr>
<tr>
<td>代码规范</td>
<td>代码规范较为灵活，可以自由组织代码结构</td>
<td>代码规范较为严格，需要按照规范组织代码结构</td>
</tr>
</tbody>
</table>
<h4 id="建议">建议：</h4>
<p>根据上述比较，我们可以得出以下选择建议：</p>
<ul>
<li>如果需要使用全面的ORM功能，或者需要使用复杂的查询构建和事务处理等功能，建议选择SQLAlchemy。</li>
<li>如果需要使用轻量级的ORM功能，或者<strong>需要快速上手和使用</strong>，建议选择sqlmodel。</li>
<li>如果需要支持多种数据库，建议两者都可以考虑使用。</li>
<li><strong>如果对文档和社区支持有较高的要求</strong>，建议选择SQLAlchemy。</li>
<li>如果对代码规范有较高的要求，建议选择sqlmodel。</li>
</ul>
</div>
<div id="MySignature" role="contentinfo">
<div style="border: 1px solid #e4e4e4 !important; font-size: 13px; padding: 5px; padding-left: 20px">
<div style="text-align: left; font-weight: bold; color: #464545; font-family: -webkit-pictograph">文中可能存在描述不正确，欢迎大神们指正补充！</div>
<div style="text-align: left; font-weight: bold; color: #0eaf48; font-family: -webkit-pictograph">感谢阅读，如果觉得对你有帮助，就在右下角点个赞吧，感谢！</div>
</div>
<p style="padding: 10px; border-top: 1px solid #f1f0f0 !important; font-size: 15px; font-family: cursive">合抱之木，生于毫末；九层之台，起于累土；千里之行，始于足下。</p><br><br>
来源：https://www.cnblogs.com/Detector/p/17458890.html
頁:
[1]