首页 > 数据库 >fastapi 使用sqlalchemy

fastapi 使用sqlalchemy

时间：2024-08-15 09:54:36浏览次数：12  
标签：comment sqlalchemy Column fastapi db User 使用 import user

一、简介

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

相关文章

  • 金蝶云星空授权及使用SDK
    网址相关1.开放平台地址：https://openapi.open.kingdee.com/ApiDoc2.第三方授权地址：https://open.kingdee.com一、金蝶云星空授权第三方1.登录金蝶云星空后台，找到第三方授权设置2.点击新增按钮，进入新增第三方系统登录授权功能页面。3.点击”获取应用ID”按钮，根据提示......
  • 使用 TypeScript 在 React JS 中进行路由
    一.介绍单页应用程序(SPA)中的路由支持在视图之间导航，而无需重新加载应用程序。ReactRouter是React应用程序中路由的标准库。本文简要概述了使用TypeScript设置路由的方法。二.设置项目创建一个新的React项目npxcreate-react-appreact-router-ts--template......
  • java使用动态链接库读取Fanuc设备，在linux环境部署时报错：FOCAS2 log file is not found
    在linux环境中，使用java调用动态链接库的方式读取Fanuc，报错“FOCAS2logfileisnotfound”解决办法linux环境使用cnc_allclibhndl3之前，需要先使用cnc_startupprocess启用并指定日志文件，否则会报错："FOCAS2logfileisnotfound"。会包含cnc_startupprocess，windows的dll库......
  • 使用 JavaScript 进行线性搜索
    一.介绍线性搜索，也称为顺序搜索，是一种用于在列表中查找特定值的简单搜索算法。它的工作原理是逐个检查列表中的每个元素，直到找到所需的值或到达列表的末尾。以下是线性搜索如何工作的逐步描述。**从头开始：**从列表的第一个元素开始。**比较各个元素：**将当前元素与目标值......
  • MPP库使用row_number()取值错误问题
    问题MPP库使用row_number()进行排序后，取指定序号值时，是随机取值了，并没有取指定序号的数据witht1as( selectrow_number()over(partitionbypidorderbycreate_timedesc)asnum,*fromt3)select*fromt1wherenum=1原因暂时还没了解，如果有哪......
  • Redis Desktop Manager(Redis可视化工具)安装及使用详细教程
    一、安装包下载直接从官网下载，官网下载链接地址：Downloads-Redis二、安装步骤2.1说明RedisDesktopManager是一款简单快速、跨平台的Redis桌面管理工具，也也被称作Redis可视化工具。支持命令控制台操作，以及常用，查询key、rename、delete等操作。2.2安装步骤2.2.1双击运......
  • C语言中水平制表符 \t 与退格键 \b 的使用方法探索
    经个人实践，C语言中使用转义序列码（\t）会输出一个8个格数的组合，当\t之前的内容达到8*n格时，后续内容出现在8*(n+1)+1格。（式中n>=1，且为整数）探索过程如下：第一次在看到某大佬的科普中提到：\b将输出位置左移一位\t表示一个tab的距离即1个大空格，相当于4个小空格......
  • 如何使用Typora写出自己的第一个博客
    markdown的使用说明一、标题语法：#这是一级标题##这是二级标题......代码：#这是一级标题##这是二级标题快捷键：Ctrl+数字:数字1-6可以快速将选中的文本调成对应级别的文本Ctrl+0:调成普通文本Ctrl+加号或者减号：对标题级别进行加减二、段落1、换行代码：1这......
  • arthas的使用入门
    官网代码仓库AlibabaJavaDiagnosticToolArthas/AlibabaJava诊断利器Arthas官方文档-中文官方文档-英文使用指导常见问题常见的现象，如下：CPU占用率高，接口的响应时延大。CPU占用率低，接口的响应时延大或者无响应。功能测试的场景下，CPU占用率超出预期。压力测试......
  • 如何在Ubuntu上使用HMCL游玩Minecraft
    想要训练一个基于计算机视觉技术的Minecraft的AI，需要在Ubuntu系统上启动Minecraft游戏本体。前置条件假设你已经在Ubuntu安装了OpenJDK或其他的JDK。你需要运行的是Minecraft的Java版。必要的下载开源的Minecraft启动器HMCL：HMCL启动器的下载链接HMCL......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99