Python框架篇(9):FastApi-SQLAlchemy集成

news2025/2/26 5:49:25

1. 介绍

ORM框架将数据库中的表（表结构）映射为面向对象的类（对象），将表中的记录（行）映射为类的实例（对象的实例），将表中的字段（列）映射为类的属性（对象的属性）。通过ORM框架，可以直接使用面向对象的方式来进行数据库操作，比如增删改查等，而不必编写复杂的SQL语句。

ORM框架的主要作用包括：

  1. 简化数据库操作： ORM 框架封装了底层数据库的操作细节，提供了高级的对象操作接口，使得数据库操作更加简单和直观。
  2. 提高开发效率： ORM 框架提供了自动化的数据库映射和对象关系管理，减少了开发人员对数据库操作的工作量，提高了开发效率。
  3. 提高代码的可维护性：通过 ORM框架，可以将数据模型和业务逻辑解耦，使得代码更加模块化和易于维护。
  4. 提高跨数据库的兼容性： ORM 框架通常具有良好的跨数据库兼容性，可以轻松地切换不同的数据库，而不必修改大量的代码。
  5. 防止SQL注入攻击： ORM框架通常会对用户输入的数据进行参数化处理，从而有效地防止了 SQL注入攻击。

2. 依赖安装

2.1 安装sqlalchemy

@注意: 虽然sqlalchemy已经升级到2.0, 但发现自动生成模型工具sqlacodegen还是基于sqlalchemy 1.4生成代码，所以这里仍然使用sqlalchemy 1.4版本 。

# 安装
$  python-learn pip install sqlalchemy==1.4.51
...
Installing collected packages: sqlalchemy
Successfully installed sqlalchemy-1.4.51

2.2 安装模型生成器

# 这里指定版本安装，本人体验的是最新版本
$ pip install sqlacodegen==3.0.0rc3

3.生成model

3.1 编写脚本

文件: bin/genmodels.sh

#!/bin/bash
# 判断参数是否为空
if [ -z "$1" ]; then
    echo -e " 使用说明: $0 connect db_type
[connect示例]:
 mysql: mysql+pymysql://用户名:密码@127.0.0.1:3306/数据库名
 postgresql: postgresql://username:password@localhost:5432/database_name
 mongodb: mongodb://username:password@localhost:27017/database_name
[db_type示例]:
  mysql、postgresql、mongodb
    "
    exit 1
fi

# 提取数据库类型
db_type=$(echo "$2" | awk -F: '{print $1}')
# 模型文件目录
model_path="app/dao/models/"

echo "db_type: $db_type"
# 生成模型文件名
output_file="$2"
case "$db_type" in
    mysql)
        output_file="${output_file}_gen.py"
        ;;
    postgresql)
        output_file="${output_file}_gen.py"
        ;;
    mongodb)
        output_file="${output_file}_gen.py"
        ;;
    *)
        echo "数据库类型只能是[mysql/postgresql/mongodb] database type: $db_type"
        exit 1
        ;;
esac

# 使用 sqlacodegen 生成模型文件
sqlacodegen "$1" > "${model_path}$output_file"
echo "Generated models file: $output_file"

3.2 运行脚本

# 赋执行权限
$ chmod 777 bin/genmodels.sh 

# 运行
$ bash bin/genmodels.sh                                                                                        
 使用说明: bin/genmodels.sh connect db_type
[connect示例]:
 mysql: mysql+pymysql://用户名:密码@127.0.0.1:3306/数据库名
 postgresql: postgresql://username:password@localhost:5432/database_name
 mongodb: mongodb://username:password@localhost:27017/database_name
[db_type示例]:
  mysql、postgresql、mongodb

3.3 生成model

@注意: 关于本次测试使用的库SQL文件在目录: static/sql/user.sql

# 执行脚本
$ bash bin/genmodels.sh mysql+pymysql://root:root@127.0.0.1:3306/test mysql
db_type: mysql
Generated models file: mysql_gen.py

运行上述命令后，会把数据库(test)中所有的表，生成对应的model，存到文件：app/dao/models/mysql_gen.py中。

4. 封装集成

4.1 添加配置

文件: .env

# -------- 数据库配置 --------
DB_DSN=mysql+pymysql://root:root@127.0.0.1:3306/test # 数据库连接
DB_ECHO_SQL=True # 使用打印SQL日志信息
DB_POOL_SIZE=5  # 连接池中的初始连接数，默认为 5
DB_MAX_OVERFLOW=10  # 连接池中允许的最大超出连接数

4.2 封装会话

文件: app/dao/base_dao.py

from sqlalchemy import create_engine
from app.config import globalAppSettings
from sqlalchemy.orm import sessionmaker
from contextlib import contextmanager

# 创建引擎
engine = create_engine(
    globalAppSettings.db_dsn,
    echo=globalAppSettings.db_echo_sql,  # 是否打印SQL
    pool_size=globalAppSettings.db_pool_size,  # 连接池的大小，指定同时在连接池中保持的数据库连接数，默认:5
    max_overflow=globalAppSettings.db_max_overflow,  # 超出连接池大小的连接数，超过这个数量的连接将被丢弃,默认: 5
)

# 封装获取会话
Session = sessionmaker(bind=engine, expire_on_commit=False)


@contextmanager
def getDatabaseSession(autoCommitByExit=True):
    """使用上下文管理资源关闭"""
    _session = Session()
    try:
        yield _session
        # 退出时，是否自动提交
        if autoCommitByExit:
            _session.commit()
    except Exception as e:
        _session.rollback()
        raise e

@注意：使用sessionmaker需要设置属性expire_on_commit=False，否则会出现报错:Instance is not bound to a Session  

4.3 封装dao

文件: app/dao/user_dao.py


from sqlalchemy import desc
from .base_dao import getDatabaseSession
from app.dao.models import YmUser


class UserQueryDao(object):
    """用户查询类dao"""

    @classmethod
    def findByPhone(cls, phone: str) -> YmUser:
        """单条查询示例"""
        with getDatabaseSession() as session:
            query = session.query(YmUser).filter(YmUser.phone == phone)
            result = query.first()
        return result

    @classmethod
    def findByPage(
        cls, page: int = 1, pageSize: int = 10, **kwargs
    ) -> (int, list[YmUser]):
        """分页查询示例"""
        with getDatabaseSession() as session:
            query = session.query(YmUser)
            # 填充具体查询条件
            for column, value in kwargs.items():
                if not hasattr(YmUser, column):
                    continue

                # 根据值类型，来组装查询条件
                if isinstance(value, tuple):
                    # 范围查询
                    query = query.filter(getattr(YmUser, column).between(*value))
                elif isinstance(value, list):
                    # in查询
                    query = query.filter(getattr(YmUser, column).in_(value))
                elif isinstance(value, str) and value.find("%") != -1:
                    # 模糊查询
                    query = query.filter(getattr(YmUser, column).like(value))
                else:
                    # 等值查询
                    query = query.filter(getattr(YmUser, column) == value)

            # 查询总条数
            total = query.count()
            # 排序分页
            offset = (page - 1) * pageSize
            query = query.order_by(desc(YmUser.id)).offset(offset).limit(pageSize)
            # 查询记录
            result = query.all()

        return total, result


class UserOperateDao(object):
    """操作用户相关dao"""

    @classmethod
    def saveUser(cls, user: YmUser) -> YmUser:
        """添加单条"""
        with getDatabaseSession(False) as session:
            session.add(user)
            session.commit()
            session.refresh(user)
        return user

    @classmethod
    def saveUserList(cls, users: list[YmUser]):
        """添加单条"""
        with getDatabaseSession() as session:
            session.bulk_save_objects(users)
        return

4.4 单元测试

文件: tests/user_query_dao_test.py

在之前的文章: Python库学习(十四):ORM框架-SQLAlchemy: https://mp.weixin.qq.com/s/y9FTKYl_Kf6opNgddUWa1g 学过SQLAlchemy的一些基本使用，这里只做简单示例演示

import unittest
from datetime import datetime

from app import dao
from app.dao import models


class UserDaoTestCase(unittest.TestCase):
    def test_findByPhone(self):
        """单条查询测试"""
        result = dao.UserQueryDao.findByPhone("17408049453")
        self.assertNotEqual(result.id, 0)

    def test_findByPage(self):
        """分页查询测试"""
        # 10, age=30, gender='male', height=(160, 180), city=['New York', 'Los Angeles']
        total, result = dao.UserQueryDao.findByPage(
            1,
            10,
            id=(10, 30),
            phone=["17804116371", "17350624789", "17654732912", "17545435626"],
            nick_name="%雨%",
        )
        self.assertNotEqual(len(result), 0)

    def test_saveUser(self):
        """单条查询测试"""
        result = dao.UserOperateDao.saveUser(
            models.YmUser(
                union_id="ui_12344343434",
                open_id="op_ksjdhjjkdhdjdhh",
                nick_name="娃哈哈",
                password="123456",
                email="test@163.com",
                phone="17600000000",
                last_login=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
                avatar="http://img-avatar.com/head-abc.jpg",
            )
        )
        print(result.id)
        self.assertNotEqual(result.id, 0)


if __name__ == "__main__":
    unittest.main()

5. 业务实例

5.1 请求流程简述

现在要实现一个简单的用户列表查询接口，下面是整个业务的简图:

image-20240510154215412
image-20240510154215412

5.2 定义控制器

文件: app/controller/user_router.py

from fastapi import APIRouter
from app.types import apiproto
from app.service import usersvc
from app import utils

router = APIRouter(prefix="/user", tags=["用户相关接口"])

@router.post("/list")
async def userList(param: apiproto.UserListRequest) -> utils.HttpResponse:
    """
    用户列表-演示
    """
    data = usersvc.UserListService.getUserList(param)
    return utils.ResponseSuccess(data)

@注意: 此处为了节省篇幅，忽略注册控制到框架的代码，具体使用可参考之前的代码。

5.3 编写servcie

文件: app/service/usersvc/user_list_svc.py

from typing import List
from app.types import apiproto
from app import dao


class UserListService:
    """用户列表"""

    @classmethod
    def getUserList(cls, queryParam: apiproto.UserListRequest) -> apiproto.UserListResponse:
        """查询用户列表"""

        # 拼凑查询信息
        queryDict = {}
        if queryParam.nick_name != "":
            queryDict["nick_name"] = f"%{queryParam.nick_name}%"
        if queryParam.phone != "":
            queryDict["phone"] = queryParam.phone

        total, result = dao.UserQueryDao.findByPage(
            queryParam.page,
            queryParam.pageSize,
            **queryDict
        )
        if total == 0:
            return apiproto.UserListResponse()

        # 格式化数据
        records_list: List[apiproto.UserDetailProto] = []

        for record in result:
            tmp = apiproto.UserDetailProto(
                id=record.id,
                union_id=record.union_id,
                open_id=record.open_id,
                nick_name=record.nick_name,
                avatar=record.avatar,
                phone=record.phone,
                email=record.email,
                last_login=record.last_login,
                status=record.status,
                delete_at=record.delete_at,
                created_at=str(record.created_at),
                updated_at=str(record.updated_at),
            )
            records_list.append(tmp)

        return apiproto.UserListResponse(record_total=total, record_list=records_list)

5.4 请求验证

5.4.1 发起请求
curl -X 'POST' \
  'http://0.0.0.0:8088/api/user/list' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "nick_name": "辉",
  "phone": "",
  "page": 1,
  "pageSize": 10
}'
5.4.2 响应结果
{
  "code": 200,
  "msg": "处理成功",
  "data": {
    "record_total": 128,
    "record_list": [
      {
        "id": 9866,
        "union_id": "ui_d6YXU4Ie3yaHnDiF7dRrl2iRWnB1Cz9gshxPkDb9xEf5f12j9X",
        "open_id": "op_KQyeQOV1b0nVv823PP4HZvsM3XBeOdq8hPiy8aK6AYcBW9aLss",
        "nick_name": "传辉宇彬胜",
        "avatar": "http://img-avatar.com/head-DveZAsPp.jpg",
        "phone": "17154750068",
        "email": "DveZAsPp@163.com",
        "last_login": "2023-12-01 11:52:28",
        "status": 1,
        "delete_at": "",
        "created_at": "2024-01-04 11:52:33",
        "updated_at": "2024-01-04 11:52:33"
      },
      {
        "id": 9711,
        "union_id": "ui_Nz2BaajGh20rUIBG23nUZT51wIxHdZbUsZMaUgPViiz11r8POn",
        "open_id": "op_lJxnCcCaj2i5wMqsGy2AE1Iqd1fRdZ0Ngv9xFuVuMBAHbwUoKc",
        "nick_name": "责た精萬辉",
        "avatar": "http://img-avatar.com/head-WOOl4Bj4.jpg",
        "phone": "15928359654",
        "email": "WOOl4Bj4@163.com",
        "last_login": "2023-11-29 11:52:27",
        "status": 1,
        "delete_at": "",
        "created_at": "2024-01-04 11:52:33",
        "updated_at": "2024-01-04 11:52:33"
      },
      省略...
    ]
  },
  "additional": {
    "time": "2024-05-10 19:01:05",
    "trace_id": "b3d74e53872c1596daf67daa379071a1"
  }
}

本文由 mdnice 多平台发布

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1674348.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Oracle21c数据库普通用户创建及授权，建表，创建存储过程、序列、触发器

Oracle21c数据库普通用户创建及授权，建表，创建存储过程、序列、触发器

一、Oracle数据库错误 ORA-65096 表示你尝试在多租户容器数据库（CDB）环境中创建一个公共用户（common user）或角色，但没有使用正确的前缀。在多租户架构中，公共用户的用户名必须以 C## 或 c## 开头。 若想…
阅读更多...
Linux|基础IO

Linux|基础IO

回顾c语言的文件操作 #include<stdio.h> int main() { FILE * fp fopen("test.txt","w"); if(fp NULL) return -1;fwrite(message,strlen(message),1,fp); fclose(fp); return 0; } 我们我们以写的方式打开 不存在则创建…
阅读更多...
Kubernetes——CNI网络组件

Kubernetes——CNI网络组件

目录 一、Kubernetes三种接口 二、Kubernetes三种网络 三、VLAN与VXLAN 1.VLAN 2.VXLAN 3.区别 3.1作用不同 3.2vxlan支持更多的二层网络 3.3已有的网络路径利用效率更高 3.4防止物理交换机Mac表耗尽 3.5相对VLAN技术&#xff0c;VXLAN技术具有以下优势 四、CNI网…
阅读更多...
爱普生M-A352加速度计受到日本气象厅认证

爱普生M-A352加速度计受到日本气象厅认证

地震一直是缠在人们头顶的乌云&#xff0c;如何能在地震发生的时候提前获悉&#xff0c;防止造成更大的经济损失&#xff0c;成为了许多企业准备解决的问题。精工爱普生公司获悉&#xff0c;东京Knowledge ForesightInc.生产的配备爱普生M-A352 高性能三轴加速度计的“Yure Mon…
阅读更多...
灌区信息化管理平台系统包含哪些内容？（全面介绍）

灌区信息化管理平台系统包含哪些内容？（全面介绍）

政策背景 2022年12月29日&#xff0c;水利部启动48处大中型灌区开展数字孪生灌区先行先试建设。 2023年2月24日&#xff0c;《2023年农村水利水电工作要点》:2023年农村水利水电工作的总体思路包括:紧盯保障国家粮食安全&#xff0c;加快推进大中型灌区现代化改造&#xff0c;…
阅读更多...
传说中的运维门户设计

传说中的运维门户设计

在IT服务管理这片广阔天地中&#xff0c;运维门户如同一位技艺高超的魔术师&#xff0c;轻轻一挥手&#xff0c;便将纷繁复杂的运维世界化繁为简&#xff0c;编织成一张便捷高效、触手可及的网络。它不仅是ITSM系统中不可或缺的一环&#xff0c;更是连接用户与技术世界的桥梁&a…
阅读更多...
mars3d实现gltf模型new mars3d.graphic.ModelPrimitive({的自定义shader

mars3d实现gltf模型new mars3d.graphic.ModelPrimitive({的自定义shader

原模型展示&#xff1a; 自定义shader效果展示&#xff1a; 运动状态下&#xff1a; 关键代码&#xff1a; const pointCloudWaveShader new Cesium.CustomShader({uniforms: {u_time: {type: Cesium.UniformType.FLOAT,value: 0}},vertexShaderText: void vertexMain(Vertex…
阅读更多...
【全开源】房屋出租出售预约系统支持微信小程序+H5+APP

【全开源】房屋出租出售预约系统支持微信小程序+H5+APP

一款基于FastAdminThinkPHPUniapp开发的房屋出租出售预约系统&#xff0c;支持小程序、H5、APP&#xff1b;包含房客、房东(高级授权)、经纪人(高级授权)三种身份。核心功能有&#xff1a;新盘销售、房屋租赁、地图找房、小区找房&#xff0c;地铁找房等方式。 特色功能&#…
阅读更多...
XML解析 之 DomFourJ解析

XML解析 之 DomFourJ解析

1&#xff0c;DomFourJ干嘛的&#xff1f; 百度一搜一大堆而且说的很繁琐&#xff0c;在这总结一句话&#xff1a; dom4j就是一个Java用来读写XML文件的API&#xff0c;而且简单又方便还好用 2&#xff0c;什么时候用&#xff1f; 不管读取什么XML文档只要你想用就用没啥限制…
阅读更多...
浅谈工业用LED面光源

浅谈工业用LED面光源

在机器视觉系统中&#xff0c;光源作为不可或缺的一部分&#xff0c;能够提高目标成像效果&#xff0c;增强检测效果。光源的选择至关重要&#xff0c;选到不合适的会影响成像及检测效果。针对不同的检测对象,不同的形状光源应运而生。我们来看看最常用的LED光源之一——面光源…
阅读更多...
『 Linux 』重定向 Redirect(万字)

『 Linux 』重定向 Redirect(万字)

文章目录 &#x1f9f8; 什么是重定向&#x1f421; 文件描述符的分配规则&#x1f421; 重定向在日常使用中的简单示例 &#x1f9f8; 实现重定向的底层机制&#x1f421; dup2()&#x1f421; 利用dup2()实现重定向 &#x1f9f8; 在自定义Shell当中添加重定向功能&#x1f4…
阅读更多...
【Neo4jJDK开箱即用的安装全流程】

【Neo4jJDK开箱即用的安装全流程】

neo4j:命令行本地访问loclhost neo4j:命令行本地访问loclhost2 neo4j操作 Neo4j桌面版数据库导出导入 Neo4j安装与配置以及JDK安装与配置教程&#xff08;超详细&#xff09; Neo4j 安装、使用教程 Neo4j安装教程 Neo4J桌面版的配置和连接Pycharm jdk-neo对应版本 JDK ORACLE中…
阅读更多...
2024/5/14 英语每日一段

2024/5/14 英语每日一段

“It is important as it suggests that possibly several populations in the world already started to include substantial amount of plants in their diet” in the period before agriculture was developed, a view contradictory to the popular one, added archeo-ge…
阅读更多...
全网最通俗易懂的vue透传

全网最通俗易懂的vue透传

概念&#xff1a; Vue的透传是指在Vue组件中&#xff0c;使用特定的语法将父组件传递的属性或事件直接传递给子组件&#xff0c;实现了通过父组件传递数据或事件&#xff0c;再传递给子组件的功能。&#xff08;传递给一个组件&#xff0c;却没有被该组件声明为 props 或 emit…
阅读更多...
2024年前一季度，国内医疗器械营收TOP10出炉！

2024年前一季度，国内医疗器械营收TOP10出炉！

随着国内医疗器械市场的不断发展&#xff0c;各大医疗器械公司的财报数据成为了投资者和行业观察者关注的焦点。近日&#xff0c;根据2024年第一季度财报数据&#xff0c;我们梳理出了中国医疗器械第一财季营收排名前十的械企&#xff0c;为大家带来深入的分析和解读。 一、营…
阅读更多...
算法课程笔记——路径相关树形DP

算法课程笔记——路径相关树形DP

算法课程笔记——路径相关树形DP #include<bits/stdc.h>usingnamespacestd; usingLL longlong; constintN 2005; vector<int>e[N],t; structasdf{vector<int> vec; LL val; }; vector<asdf>w[N]; LL dp[N]; intn,m,k,dep[N]{1},f[N]; voiddfs(in…
阅读更多...
图生视频，Stable Diffusion WebUI Forge内置SVD了！

图生视频，Stable Diffusion WebUI Forge内置SVD了！

在 Stable Diffusion WebUI Forge 版本中内置了一个SVD插件&#xff0c;也就是 Stable Video Diffusion&#xff08;稳定视频扩散&#xff09;&#xff0c;之前我介绍过这个工具的使用方法&#xff1a;图片生成视频&#xff08;独立部署SVD) 但是当时还不能集成到Stable Diffu…
阅读更多...
【代码阅读】SalsaNext

【代码阅读】SalsaNext

最近在找轻量级的语义分割模型&#xff0c;SalsaNext作为一个很经典的语义分割网络&#xff0c;在服务器的2080上面能够达到30毫秒一帧左右的推理速度&#xff0c;但是其网络本身提出的时间比较久远&#xff0c;后处理的部分使用的依然是最经典的knn&#xff0c;fidnet的后处理…
阅读更多...
【已解决】attributeerror: ‘FreeTypeFont‘ object has no attribute ‘getsize‘

【已解决】attributeerror: ‘FreeTypeFont‘ object has no attribute ‘getsize‘

&#x1f60e; 作者介绍&#xff1a;我是程序员行者孙&#xff0c;一个热爱分享技术的制能工人。计算机本硕&#xff0c;人工制能研究生。公众号&#xff1a;AI Sun&#xff0c;视频号&#xff1a;AI-行者Sun &#x1f388; 本文专栏&#xff1a;本文收录于《AI实战中的各种bug…
阅读更多...
Spring Cloud 概述及项目创建

Spring Cloud 概述及项目创建

本篇主要介绍什么是Spring Cloud&#xff0c;以及Spring Cloud工程的创建 目录 一、什么是微服务&#xff1f; 集群 分布式 微服务 二、Spring Cloud 什么是Spring Cloud Spring Cloud 版本 Spring Cloud实现方案 Spring Cloud 工程创建 创建父工程 创建子工程 一、…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023