一、简介
fastapi 常见的orm框架有以下几种:
SQLAlchemy:这个比较常见,之前用flask开发web框架也用的SQLAlchemy。
SQLModel:网上说是最适合fastapi的orm框架,官方也推荐这个,后续应该会发展不错,目前没有去踩坑。
tortoise-orm:django的异步orm框架,与fastapi也兼容,没用过不做评价。
这里介绍下SQLAlchemy在fastapi框架的使用。
二、安装SQLAlchemy
pip install sqlalchemy
如果链接的是mysql,需要安装pymysql
pip install pymysql
三、配置SQLAlchemy
项目目录:
SessionLocal 类
它是一个本地线程存储(thread-local storage)的单例类,用来创建数据库会话,它是由工厂函数 sessionmaker 创建的。简单来说,SessionLocal 类的主要作用是为每个请求创建一个数据库会话,并且确保这个会话在整个请求期间都是唯一的。
declarative_base()
declarative_base() 是 SQLAlchemy 中提供的一个函数,用于创建一个基类,然后通过继承这个基类来定义数据表模型。它可以让我们更加方便地定义数据表模型,而不需要关注底层的SQL语句。具体作用:
- 自动创建对应的数据表:我们定义了数据表模型之后,可以调用 create_all() 方法来创建对应的数据表。
- 自动映射数据表和类属性:我们只需要定义类属性,SQLAlchemy 可以自动将这些属性映射到对应的数据表字段。
- 提供了更加易读易懂的代码:使用 declarative_base() 可以让我们更加方便地定义类,使代码更加清晰易读。
1、我们创建一个 plugin/plugin_sqlalchemy.py 文件,用来初始化 SQLalchemy 引擎
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/fastapi?charset=utf8mb4"
POOL_SIZE = 20
# SQLALCHEMY_DATABASE_URL = "postgresql://root:123456@postgresserver/db"
#创建一个 SQLAlchemy的“引擎”
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
pool_size=POOL_SIZE,
)
# SessionLocal该类的每个实例将是一个数据库会话。该类本身还不是数据库会话。
# 一旦我们创建了SessionLocal该类的实例,该实例将成为实际的数据库会话。
# 要创建SessionLocal类,请使用函数sessionmaker,sessionmaker是一个工厂函数,返回一个配置好的类
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 实例化SessionLocal类
db = SessionLocal()
#我们将用这个类继承,来创建每个数据库模型或类(ORM 模型)
Base = declarative_base()
2、在models层创建 baseModel.py 文件,定义一个基础数据表模型类,之后的业务数据表模型都继承 baseModel 类。
from plugin.pulgin_sqlalchamy import Base
from sqlalchemy import Column,Integer,DATETIME
from datetime import datetime
class baseModel(Base):
# 定义为抽象类,只能被继承,不能实例化
__abstract__ = True
# 默认字段
id = Column(Integer, primary_key=True, autoincrement=True, comment="主键ID")
create_user = Column(Integer, default=0, comment="创建人")
create_time = Column(DATETIME, default=datetime.now, comment="创建时间")
update_user = Column(Integer, default=0, comment="更新人")
update_time = Column(DATETIME, default=datetime.now, comment="更新时间")
is_delete = Column(Integer, default=0, comment="删除标识:0-正常 1-已删除")
3、创建业务数据表模型 models/cms/user.py
from models.baseModel import baseModel
from sqlalchemy import Column,Integer,String
class User(baseModel):
__tablename__ = "user" # 数据库表名
username = Column(String(20), nullable=False, unique=True, comment="用户姓名")
password = Column(String(20), nullable=False, comment="密码")
nickname = Column(String(50),nullable=True, comment="昵称")
email = Column(String(50), nullable=True, comment="电子邮箱")
4、models/__init__.py 自己封装一个方法,使用create_all() 创建数据库表。
from plugin.pulgin_sqlalchamy import engine,Base
def register_database():
# 预先创建数据表
from .cms import user
Base.metadata.create_all(bind=engine)
5、api/__inti__.py 定义 create_app 方法,注册路由,注册插件,加载配置等。
from fastapi import FastAPI,Request
import time
from routers import register_router
from models import register_database
app = FastAPI()
def creat_app():
# 注册路由
register_router(app)
# 注册数据库
register_database()
@app.middleware("http")
# 定义中间件功能
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = round((time.time() - start_time) * 1000, 2)
response.headers["X-Process-Time"] = f'{str(process_time)}ms'
return response
return app
6、然后在main.py调用 create_app 方法,初始化 fastapi 框架。
from api import creat_app
import uvicorn
app = creat_app()
if __name__ == "__main__":
uvicorn.run(app='main:app', host="0.0.0.0", log_config='./uvicorn_config.json', port=8000)
服务启动后,可以看到数据库新创建了一张user表
四、API接口中操作数据库
1、往数据库插入一条数据,并且验证数据类型,get_db() 方法处理了数据库会话异常后,finally 退出会话。
from fastapi import APIRouter,Depends
from models.cms.user import User
from sqlalchemy.orm import Session
from schemas.cms.user import userSchema
import logging
from plugin.pulgin_sqlalchamy import db
log = logging.getLogger('uvicorn')
user = APIRouter()
def get_db():
try:
yield db
finally:
db.close()
@user.get('/test')
async def test():
return {'message': 'test'}
@user.post('/dev')
async def dev(data: userSchema, db: Session = Depends(get_db)):
db_user = User(username = data.username,password = data.password )
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
2、数据验证,新建 schemas/cms/user.py 文件,定义每个接收参数字段的类型。Field 是一个功能很强大的方法,可以校验长度,甚至可以使用正则去检查参数是否符合标准。
from pydantic import BaseModel,Field
from typing import Optional,Union
class userSchema(BaseModel):
# 数据校验username长度最小2位,最大10位,正则匹配小写字母
username: str = Field(description='用户名',min_length=2,max_length=10,regex=r'^[a-z]{2,10}$')
password: str
nickname: Optional[str] = None
email: Optional[str] = None
服务启动后,swagger 文档会展示传参类型,schema定义哪些是必传参数、类型和长度。传参不合符规则,状态码 code 会报 422:
五、外键和外键约束
通过 ForeignKey 设置 user 表的 group_id 作为 group 表的外键。
from models.baseModel import baseModel
from sqlalchemy import Column,Integer,String,ForeignKey
class User(baseModel):
__tablename__ = "user" # 数据库表名
username = Column(String(20), nullable=False, unique=True, comment="用户姓名")
password = Column(String(20), nullable=False, comment="密码")
nickname = Column(String(50),nullable=True, comment="昵称")
email = Column(String(50), nullable=True, comment="电子邮箱")
# 设置外键
group_id = Column(Integer,ForeignKey('group.id'))
class Group(baseModel):
__tablename__ = "group" # 用户组
name = Column(String(20), nullable=False,unique=True, comment='用户组')
info = Column(String(20), nullable=True,comment='描述')
level = Column(nullable=True,comment='分组级别')
外键约束分类
RESTRICT:这是默认选项。当尝试删除父表中的数据时,如果子表中存在与之关联的数据,那么这个删除操作将会被阻止。也就是说,只有当没有任何子表行与父表行关联时,才能删除父表中的行。
NO ACTION:在 MySQL 中,这个选项的行为与 RESTRICT 选项相同。也就是说,如果子表中存在与父表行关联的行,那么尝试删除父表中的行将会被阻止。
CASCADE:这个选项表示级联删除。当你删除父表中的行时,所有在子表中与之关联的行也会被自动删除。这种选项需要谨慎使用,因为它可能会导致大量的数据被删除。
SET NULL:当父表中的行被删除时,这个选项会将子表中所有与之关联的行的外键列设置为 NULL。这意味着,子表中的这些行不再与父表中的任何行关联。注意,为了使用这个选项,子表的外键列必须允许 NULL 值。
from models.baseModel import baseModel
from sqlalchemy import Column,Integer,String,ForeignKey
class User(baseModel):
__tablename__ = "user" # 数据库表名
username = Column(String(20), nullable=False, unique=True, comment="用户姓名")
password = Column(String(20), nullable=False, comment="密码")
nickname = Column(String(50),nullable=True, comment="昵称")
email = Column(String(50), nullable=True, comment="电子邮箱")
# 设置外键约束
group_id = Column(Integer, ForeignKey('group.id', ondelete='SET NULL'))
class Group(baseModel):
__tablename__ = "group" # 用户组
name = Column(String(20), nullable=False,unique=True, comment='用户组')
info = Column(String(20), nullable=True,comment='描述')
level = Column(nullable=True,comment='分组级别')
六、CRUD
上文我们构建db对象:所有和数据库的ORM操作都必须通过一个db的会话对象来实现:
1、create data
# User是数据表模型
# add 单条数据
db_user = User(username = "Tom",password = "1234567" )
db.add(db_user)
db.commit()
#add 批量数据
db_user1 = User(username = "Tom",password = "1234567" )
db_user2 = User(username = "Jack",password = "123456" )
db.add_all([db_user1,db_user2])
db.commit()
2、read data
# 搜索user表第一条数据
result = db.query(User).first()
# 搜索user表所有数据
result = db.query(User).all()
# filter过滤条件搜索
result = db.query(User).filter(User.username=='admin').first()
# filter_by过滤条件搜索
result = db.query(User).filter_by(username='admin').all()
# order_by,排序倒序desc(),正序asc()
result = db.query(User).order_by(User.create_time.desc()).all()
# group_by,查看使用相同昵称的用户有几个
from sqlalchemy import func
result = db.query(User.nickname,func.count(User.username).label('total_nickname')).group_by(User.nickname).all()
# 搜索不等于
result = db.query(User).filter(User.username != 'admin')
# like
result = db.query(User).filter(User.username.like('%ad%'))
# in
result = db.query(User).filter(User.username.in_(['root','admin','jack']))
# 同时,in也可以作用于一个Query
result = db.query(User).filter(User.username.in_(db.query(User.username).filter(User.username.like('%ad%'))))
# not in
result = db.query(User).filter(~User.username.in_(['root','admin','jack']))
# and
result = db.query(User).filter(User.username == 'admin', User.nickname == '哈哈')
# 或者是通过多次filter操作
result = db.query(User).filter(User.username == 'admin').filter(User.nickname == '哈哈')
3、update data
# 修改对象:首先从数据库中查找对象,然后将这条数据修改为你想要的数据,最后做commit操作就可以修改数据了
user = db.query(User).first()
user.username = 'lucy'
db.commit()
4、delete data
# 删除对象:将需要删除的数据从数据库中查找出来,然后使用`db.delete`方法将这条数据从db中删除,最后做commit操作就可以了
user = db.query(User).first()
db.delete(user)
db.commit()
标签:comment,sqlalchemy,Column,fastapi,db,User,使用,import,user From: https://www.cnblogs.com/shenh/p/18346981