python-27-Python ORM系列之彻底搞明白ORM概念，对ORM进行封装结合FastAPI实现数据库的增删改查，联表查询等接口

python-27-Python ORM系列之彻底搞明白ORM概念，对ORM进行封装结合FastAPI实现数据库的增删改查，联表查询等接口

一.简介

在Python基础系列ORM部分为大家介绍了如何搭建MySQL数据和MySQL一些访问配置，同时也介绍了pymysql库的封装来实现对数据库的增删改查功能，但是截止以上都没有实现ORM，好菜已经上来 这一篇文章来帮助大家彻底掌握ORM框架，并且对ORM进行封装，利用ORM和FASTAPI实现数据库的增删改查，抱歉了，这篇文章设置了VIP，这段时间天天在码文章，创作收入将给我带来更大的动力，也给大家带来更好的文章，开始我们今天的日拱一卒！

二.安装pymysql、sqlalchemy、fastapi、uvicorn

#使用豆瓣源安装 提升安装速度
pip install pymysql -i http://pypi.doubanio.com/simple --trusted-host pypi.doubanio.com
pip install sqlalchemy -i http://pypi.doubanio.com/simple --trusted-host pypi.doubanio.com
pip install fastapi -i http://pypi.doubanio.com/simple --trusted-host pypi.doubanio.com
pip install uvicorn -i http://pypi.doubanio.com/simple --trusted-host pypi.doubanio.com

pymysql：Python 实现的 MySQL 数据库连接库

本文示例是调用MySql数据库，如果需要切换其他数据库，可以下载对应插件和改变ORM对应的连接字符串即可

MySQL-Python
mysql+mysqldb://:@[:]/

pymysql
mysql+pymysql://:@/[?]

MySQL-Connector
mysql+mysqlconnector://:@[:]/

cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value…]

SQLAlchemy：Python ORM（对象关系映射）工具

fastapi：FastAPI 是一个现代的、快速的 web 框架，用于构建 API

Uvicorn：Uvicorn 是一个高性能的 ASGI 服务器，适用于运行 FastAPI 和其他支持 ASGI 的 Python web 框架。

三.ORM基础介绍

1.首先我们需要sqlalchemy 多个不同模块和库,常用的引入如下

from sqlalchemy import create_engine,Column, Integer, String, ForeignKey,DateTime,Boolean,Text
from sqlalchemy.ext.declarative import declarative_base #orm基类
from sqlalchemy.orm import sessionmaker,relationship,joinedload

create_engine：创建数据库连接引擎。

Column, Integer, String, ForeignKey, DateTime, Boolean, Text：用来定义数据库表的字段类型和约束。

declarative_base：创建 ORM 基类，供模型类继承。

sessionmaker：创建数据库会话，执行数据库操作。

relationship：定义表之间的关系。

joinedload：优化查询，避免 N+1 查询问题。

四.构建数据库连接字符串的配置文件

4.1创建配置文件db_config.ini

在当前项目中创建config文件夹，在文件夹下db_config.ini文件，作为我们的数据库配置文件，内容如下：

[localdb]
host     = localhost
user     = root
password = yourpsw
port     = 3306
database = myurlhub

4.2创建获取配置文件工具包

在当前项目中创建utils文件夹，新建__init__.py文件，里面不用写任何东西，在utils文件夹下新建config_helper.py文件，用来帮助我们获取配置文件信息，文件内容如下：

import configparserclass Config(object):def __init__(self, filename, encoding='UTF-8'):# 声明配置类对象self.config = configparser.ConfigParser()# 读取配置文件self.config.read(filename, encoding)def get_value(self, section, option):"""获取 value"""value = self.config.get(section, option)return valuedef get_items(self, section):value = dict(self.config.items(section))return value# 读取配置文件
# config.read(fileName, 'GBK')
# print(f"sections: {config.sections()}")
# print(f"options: {config.options('localdb')}")
# print(f"items: {config.items('localdb')}")

我们的整体结构文件：

在这里插入图片描述

五.读取配置文件数据 构建数据库连接引擎

在当前项目文件夹utils中 新建orm_helper.py文件作为 ORM 帮助类库，我们需要根据配置文件来构建连接字符串，从而创建我们的ORM据库连接引擎 engine = create_engine(mysql_conn_str)

orm_helper.py 目前的代码：

#orm_helper.py
from sqlalchemy import create_engine,Column, Integer, String, ForeignKey,DateTime,Boolean,Text
from sqlalchemy.ext.declarative import declarative_base #orm基类
from sqlalchemy.orm import sessionmaker,relationship,joinedload
from utils import config_helper
import os
from datetime import datetime
import uuid
'''
数据库连接拼接串语法：
'''
# 加载数据库配置
__config_file_path = os.path.join(os.getcwd(),'config\db_config.ini') 
__config = config_helper.Config(__config_file_path,'UTF-8')
__localdb = __config.get_items('localdb')
mysql_conn_str = f"mysql+pymysql://{__localdb['user']}:{__localdb['password']}@{__localdb['host']}:{__localdb['port']}/{__localdb['database']}"engine = create_engine(mysql_conn_str)

六.创建 ORM 基类，公共基类，用来定义数据库表的字段类型和约束的类

创建ORM模型的基类，公共字段基类，用于一键生成数据库对应的表字段

  1. 实现ORM基类：后续用来定义数据库表的字段类型和约束的类都要继承ORM的基类

    我们本次 要创建2个数据库表 都要继承ORM Base基类CommonFieldsMixin公共字段类，CommonFieldsMixin公共字段类，所有表都有的字段，不需要额外实现，直接继承实现

    Base = declarative_base() 
# 我们本次 要创建2个数据库表 都要继承ORM Base基类 CommonFieldsMixin为公共字段类，为所有表都有的字段，不需要额外实现
# 创建用户表
#class MUH_User(Base,CommonFieldsMixin):#创建用户登录密码表
#class MUH_User_Psw(Base,CommonFieldsMixin):

  2. 实现与数据库关联的ORM表的类：这些类后续为ORM操作的对象，我们就可像操作对象一样操作这2个类与数据库进行关联，而不需要关心SQL实现

    因为我们需要实现用户注册，故要创建2张表 用户表 和 用户密码表，一张保存用户基础信息，一张保存用户密码表

    注意有几个细节：

    1. 创建公共基类CommonFieldsMixin，让所有的表都自动包含公共基类字段

      class MUH_User(Base,CommonFieldsMixin):
class MUH_User_Psw(Base,CommonFieldsMixin)

      其中Base 是ORM基类 CommonFieldsMixin为公共基类

      ORM类定义说明：

      create_date = Column(DateTime, default=datetime.now())

      • create_date:对应数据库表的列名

      • Column：表示列

      • DateTime：数据类型，注意数据库表的类型都是基础类型，千万别搞个列表，字典，元组作为表的类型，如果非要用请转换为JSON string类型

      • default：默认值

      • ORM中常见类型约束及其他约束整理如下

        ORM表中列的类型说明举例
        Integer整数类型的字段age = Column(Integer)
        String(length)描述变长字符串类型的字段name = Column(String(50))
        Text用于存储较长文本description = Column(Text)
        Boolean描述布尔值类型的字段is_active = Column(Boolean, default=True)
        Float描述浮点数类型的字段price = Column(Float)
        Numeric(precision, scale)精确数字类型，适用于需要高精度的小数amount = Column(Numeric(10, 2)) #精度为 10 位，保留 2 位小数的数字
        Date描述日期类型字段，不包含时间部分birthdate = Column(Date)
        DateTime描述日期和时间类型字段created_at = Column(DateTime)
        Time描述时间类型字段，不包含日期部分clock_in = Column(Time)
        Interval描述时间间隔字段，用于存储持续时间（例如，2 days, 5 hours）duration = Column(Interval)
        UUID用于存储 UUID（通用唯一标识符）user_id = Column(UUID(as_uuid=True), default=uuid.uuid4)
        LargeBinary（不建议使用）描述二进制数据字段，适用于存储大块二进制数据，如文件或图像。image = Column(LargeBinary)
        Enum用于存储枚举类型字段，通常用于有限的固定选项（如性别、状态等）size = Column(Enum(‘small’, ‘medium’, ‘large’, name=‘size’))
        JSON用于存储 JSON 格式的数据preferences = Column(JSON)
        ForeignKey用于指定外键约束，表示该列是另一个表的外键user_id = Column(Integer, ForeignKey(‘users.id’))
        primary_key定义表的主键id = Column(String(50), primary_key=True, default=str(uuid.uuid4()))
        nullable是否可为nullname = Column(String(100), nullable=False)
        default默认值id = Column(String(50), primary_key=True, default=str(uuid.uuid4()))
        onupdate当更新默认值onupdate=datetime.utcnow

      创建公共基类

      #orm_helper.py
#公共基类
# 定义公共字段 Mixin 类
class CommonFieldsMixin:create_date = Column(DateTime, default=datetime.now())create_id = Column(String(50))update_date = Column(DateTime, default=datetime.now(), onupdate=datetime.now)update_id = Column(String(50))visible = Column(Boolean, default=True)descriptions = Column(Text)

    2. 所有的id 采用guid字符串来实现，故引入import uuid，

      默认自动生成id = Column(String(50), primary_key=True,default=uuid.uuid4())

      #orm_helper.py
# 创建用户表
class MUH_User(Base,CommonFieldsMixin):__tablename__ = "muh_user"id = Column(String(50), primary_key=True,default=str(uuid.uuid4()))user_name = Column(String(20))user_gender = Column(String(2))user_age = Column(Integer)user_imgs =Column(String(255))def __repr__(self):# 自动列出所有字段attrs = {column.name: getattr(self, column.name) for column in self.__table__.columns}return f"<{self.__class__.__name__}({', '.join(f'{key}={value!r}' for key, value in attrs.items())})>"#创建用户登录密码表
class MUH_User_Psw(Base,CommonFieldsMixin):__tablename__ = "muh_user_psw"id = Column(String(50), primary_key=True,default=str(uuid.uuid4())) user_id = Column(String(50))user_psw = Column(String(255))def __repr__(self):# 自动列出所有字段attrs = {column.name: getattr(self, column.name) for column in self.__table__.columns}return f"<{self.__class__.__name__}({', '.join(f'{key}={value!r}' for key, value in attrs.items())})>"

    3. def __repr__(self)是用来查询时候自动转换为对应类的实例，采用固定写法，大家照着写就行

      以下为def __repr__(self)的演变：

      #`def __repr__(self)`原本的实现方式
#格式1:def __repr__(self):return f"<MUH_User(id='{self.id}', name='{self.name}', gender='{self.gender}', grade='{self.grade}', age='{self.age}', " \f"user_img='{self.user_img}', create_date='{self.create_date}', create_id='{self.create_id}', " \f"update_date='{self.update_date}', update_id='{self.update_id}', visible='{self.visible}', descriptions='{self.descriptions}')>"
#格式2：            def __repr__(self):return "<MUH_User_Psw(id='%s', user_id='%s',user_psw='%s',carete_date='%s',carete_id='%s',update_date='%s',update_id='%s',visable='%s',descrptions='%s')>" % (self.id, self.user_id, self.user_psw, self.carete_date, self.carete_id, self.update_date,self.update_id,self.visable,self.descrptions)
#格式3：def __repr__(self):# 自动列出所有字段attrs = {column.name: getattr(self, column.name) for column in self.__table__.columns}return f"<{self.__class__.__name__}({', '.join(f'{key}={value!r}' for key, value in attrs.items())})>"

七.创建 ORM 帮助类库

其实定义以上我们已经实现了ORM，但是好的程序员在使用前会创建自己的帮助类库

在orm_helper.py接着实现 ORM 帮助类库，代码如下：

#orm_helper.pyclass OrmManager(object):def __init__(self):self.__make_session()def create_table(self):try:Base.metadata.create_all(engine)  #这个语句可以创建所有继承了Base的子类的orm数据库表except Exception as e:print(f"create_table has error with {e}")def __make_session(self):Session = sessionmaker(bind=engine)  #工厂模式，返回类self.__session = Session()def fetch_session(self):return self.__sessiondef add_records(self, objs):"""插入记录:param objs: [obj1, obj2] 或 单个 obj"""try:if isinstance(objs, list):self.__session.add_all(objs)else:self.__session.add(objs)self.__session.commit()except Exception as e:self.__session.rollback()print(f"Error adding records: {e}")def delete_records(self, objs):"""删除记录:param objs: [obj1, obj2] 或 单个 obj"""try:if isinstance(objs, list):for item in objs:self.__session.delete(item)else:self.__session.delete(objs)self.__session.commit()except Exception as e:self.__session.rollback()print(f"Error deleting records: {e}")def query_records(self, model_class, filters=None, **kwargs): #cls 表示传入类"""查询记录:param model_class: ORM类模型:param filters: 过滤条件，字典形式:param kwargs: 其他查询条件，如分页，排序等:return: 查询结果:调用方式：# 查询名字为 'John' 的用户filters = {'name': 'John'}users = orm_manager.query_records(MUH_User, filters=filters)# 分页查询，第 2 页，每页 10 条记录page = 2per_page = 10users = orm_manager.query_records(MUH_User, filters=None, offset=(page-1) * per_page, limit=per_page)# 查询所有用户并按年龄升序排序users = orm_manager.query_records(MUH_User, filters=None, order_by=[MUH_User.age])# 查询性别为 'Male' 且年龄大于 30 的用户，按年龄降序排列，显示前 5 条记录filters = {'gender': 'Male'}users = orm_manager.query_records(MUH_User, filters=filters, age=30, order_by=[MUH_User.age.desc()], limit=5)# 查询年龄为 25 或 30 的用户filters = {'age': [25, 30]}users = orm_manager.query_records(MUH_User, filters=filters)# 查询 id 为 '123' 的用户及其密码from sqlalchemy.orm import joinedloadusers_with_psw = orm_manager.query_records(MUH_User, filters={'id': '123'}, join=MUH_User.password)"""query = self.__session.query(model_class)if filters:query = query.filter_by(**filters)for key, value in kwargs.items():if hasattr(model_class, key):query = query.filter(getattr(model_class, key) == value)return query.all()def query(self, *cls):'''查询记录直接返回query'''if  cls :return self.__session.query(*cls)else:return []

说明：

  1. self.__make_session()，用来创建orm的session，这个就是ORM用来操作数据库的session 很重要
    def __make_session(self):Session = sessionmaker(bind=engine)  #工厂模式，返回类self.__session = Session()
  1. 为了让我们的帮助类库更灵活，我们通过fetch_session 来返回session，方便我们以ORM原始方式来操作
    def fetch_session(self):return self.__session

  1. 增删查方法

    def add_records(self, objs):新增数据方法

    def delete_records(self, objs):删除方法

    def query_records：查询方法

  2. 为了让我们的查询更灵活，我们定义查询基本方法def query(self, *cls)，方便我们以ORM原始方式来操作

  3. 一键生成数据库中的表函数

        def create_table(self):try:Base.metadata.create_all(engine)  #这个语句可以创建所有继承了Base的子类的orm数据库表except Exception as e:print(f"create_table has error with {e}")

八.orm_helper.py完整代码：

#orm_helper.py
from sqlalchemy import create_engine,Column, Integer, String, ForeignKey,DateTime,Boolean,Text
from sqlalchemy.ext.declarative import declarative_base #orm基类
from sqlalchemy.orm import sessionmaker,relationship,joinedload
from utils import config_helper
import os
from datetime import datetime
import uuid
'''
数据库连接拼接串语法：
'''
# 加载数据库配置
__config_file_path = os.path.join(os.getcwd(),'config\db_config.ini') 
__config = config_helper.Config(__config_file_path,'UTF-8')
__localdb = __config.get_items('localdb')
mysql_conn_str = f"mysql+pymysql://{__localdb['user']}:{__localdb['password']}@{__localdb['host']}:{__localdb['port']}/{__localdb['database']}"engine = create_engine(mysql_conn_str)
Base = declarative_base()#公共基类
# 定义公共字段 Mixin 类
class CommonFieldsMixin:create_at = Column(DateTime, default=datetime.now())create_id = Column(String(50))update_at = Column(DateTime, default=datetime.now(), onupdate=datetime.now)update_id = Column(String(50))visible = Column(Boolean, default=True)descriptions = Column(Text)
# 创建用户表
class MUH_User(Base,CommonFieldsMixin):__tablename__ = "muh_user"id = Column(String(50), primary_key=True,default=str(uuid.uuid4()))user_name = Column(String(20))user_gender = Column(String(2))user_age = Column(Integer)user_imgs =Column(String(255))def __repr__(self):# 自动列出所有字段attrs = {column.name: getattr(self, column.name) for column in self.__table__.columns}return f"<{self.__class__.__name__}({', '.join(f'{key}={value!r}' for key, value in attrs.items())})>"#创建用户登录密码表
class MUH_User_Psw(Base,CommonFieldsMixin):__tablename__ = "muh_user_psw"id = Column(String(50), primary_key=True,default=str(uuid.uuid4())) user_id = Column(String(50))user_psw = Column(String(255))def __repr__(self):# 自动列出所有字段attrs = {column.name: getattr(self, column.name) for column in self.__table__.columns}return f"<{self.__class__.__name__}({', '.join(f'{key}={value!r}' for key, value in attrs.items())})>"class OrmManager(object):def __init__(self):self.__make_session()def create_table(self):try:Base.metadata.create_all(engine)  #这个语句可以创建所有继承了Base的子类的orm数据库表except Exception as e:print(f"create_table has error with {e}")def __make_session(self):Session = sessionmaker(bind=engine)  #工厂模式，返回类self.__session = Session()def fetch_session(self):return self.__sessiondef add_records(self, objs):"""插入记录:param objs: [obj1, obj2] 或 单个 obj"""try:if isinstance(objs, list):self.__session.add_all(objs)else:self.__session.add(objs)self.__session.commit()except Exception as e:self.__session.rollback()print(f"Error adding records: {e}")def delete_records(self, objs):"""删除记录:param objs: [obj1, obj2] 或 单个 obj"""try:if isinstance(objs, list):for item in objs:self.__session.delete(item)else:self.__session.delete(objs)self.__session.commit()except Exception as e:self.__session.rollback()print(f"Error deleting records: {e}")def query_records(self, model_class, filters=None, **kwargs): #cls 表示传入类"""查询记录:param model_class: ORM类模型:param filters: 过滤条件，字典形式:param kwargs: 其他查询条件，如分页，排序等:return: 查询结果:调用方式：# 查询名字为 'John' 的用户filters = {'name': 'John'}users = orm_manager.query_records(MUH_User, filters=filters)# 分页查询，第 2 页，每页 10 条记录page = 2per_page = 10users = orm_manager.query_records(MUH_User, filters=None, offset=(page-1) * per_page, limit=per_page)# 查询所有用户并按年龄升序排序users = orm_manager.query_records(MUH_User, filters=None, order_by=[MUH_User.age])# 查询性别为 'Male' 且年龄大于 30 的用户，按年龄降序排列，显示前 5 条记录filters = {'gender': 'Male'}users = orm_manager.query_records(MUH_User, filters=filters, age=30, order_by=[MUH_User.age.desc()], limit=5)# 查询年龄为 25 或 30 的用户filters = {'age': [25, 30]}users = orm_manager.query_records(MUH_User, filters=filters)# 查询 id 为 '123' 的用户及其密码from sqlalchemy.orm import joinedloadusers_with_psw = orm_manager.query_records(MUH_User, filters={'id': '123'}, join=MUH_User.password)"""query = self.__session.query(model_class)if filters:query = query.filter_by(**filters)for key, value in kwargs.items():if hasattr(model_class, key):query = query.filter(getattr(model_class, key) == value)return query.all()def query(self, *cls):'''查询记录直接返回query'''if  cls :return self.__session.query(*cls)else:return []

项目结构：

在这里插入图片描述

九.使用ORM帮助类库，通过FastAPI接口方式实现

1.FastAPI的构建方式

下载依赖fastapi 、uvicorn

pip install fastapi uvicorn -i http://pypi.doubanio.com/simple --trusted-host pypi.doubanio.com

2.FastAPI的基本创建API路由格式

from fastapi import FastAPIapp = FastAPI()  # 创建 FastAPI 应用实例@app.get("/")
def read_root():return {"message": "Hello, World!"}

3.vscode 启动方式

vscode 终端—>新建终端 运行uvicorn main:app --reload --port 8081,当出现以下表示运行成功！端口可以自由切换

uvicorn main:app --reload --host 192.168.0.1 --port 8081

main:指定运行的文件名 main.py

app: 是 FastAPI 应用对象

reload: 是在代码修改后自动重载应用

host :服务器地址

port ：指定端口号

(my_venv) PS F:\开发源码\python_demo_06> uvicorn main:app --reload --port 8081
INFO:     Will watch for changes in these directories: ['F:\\开发源码\\python_demo_06']
INFO:     Uvicorn running on http://127.0.0.1:8081 (Press CTRL+C to quit)
INFO:     Started reloader process [25572] using statreload
INFO:     Started server process [13932]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

FastAPI 会自动为你的 API 生成 Swagger UI 和 ReDoc 文档：

http://127.0.0.1:8081/docs

http://127.0.0.1:8081/redoc

  • 访问 http://127.0.0.1:8081/docs，你将看到 Swagger UI 文档。

在这里插入图片描述

  • 访问 http://127.0.0.1:8081/redoc，你将看到 ReDoc 文档。

在这里插入图片描述

十.FastAPI+ORM实现代码一键生成数据库中对应的表

  1. 一键生成数据库中对应的表

    # main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from utils import orm_helper as orm
from datetime import datetime
import uuid
from sqlalchemy.orm import joinedload
from sqlalchemy import joinapp = FastAPI()# 1. 创建表
@app.get("/create_table")
def create_table():db = orm.OrmManager()db.create_table()return {"code": 200, "success": "Tables created successfully", "data": ""}

直接直接调用http://127.0.0.1:8081/docs提供的API来调试我们的接口

在这里插入图片描述

执行后的数据库，可见我们的表和字段都创建成功！

在这里插入图片描述

十一.FastAPI+ORM实现新增用户和密码

代码如下:

# main.py
# 2. 新增用户和密码
class UserCreate(BaseModel):user_name: struser_gender: struser_age: intuser_imgs: Optional[str] = Noneuser_psw: str@app.post("/create_user")
def create_user(user: UserCreate):db = orm.OrmManager()# 创建用户new_user = orm.MUH_User(id=str(uuid.uuid4()),user_name=user.user_name,user_gender=user.user_gender,      user_age=user.user_age,user_imgs=user.user_imgs)# 创建密码new_psw = orm.MUH_User_Psw(id=str(uuid.uuid4()),user_id=new_user.id,user_psw=user.user_psw)# 插入到数据库db.add_records([new_user, new_psw])return {"code": 200, "success": "User and password created successfully"}

调用接口：

在这里插入图片描述

数据库写入信息：

在这里插入图片描述

十二.查询用户信息

代码如下：

# main.py
# 3. 查询用户信息
@app.get("/get_user/{user_id}")
def get_user(user_id: str):db = orm.OrmManager()# 查询用户和密码user = db.query_records(orm.MUH_User, filters={"id": user_id})password = db.query_records(orm.MUH_User_Psw, filters={"user_id": user_id})if user and password:return {"user": user[0], "password": password[0].user_psw}else:raise HTTPException(status_code=404, detail="User not found")

查询结果：

在这里插入图片描述

十三.更新用户信息

代码如下：

# main.py
# 4. 更新用户信息
class UserUpdate(BaseModel):id:Optional[str] = Noneuser_name: Optional[str] = None@app.put("/update_user/{user_id}")
def update_user(user: UserUpdate):try:db = orm.OrmManager()session = db.fetch_session()user_to_update  = session.query(orm.MUH_User).filter_by(id = user.id).first()print(user_to_update )user_to_update .user_name = user.user_nameif not user_to_update:raise HTTPException(status_code=404, detail="User not found")# 更新用户字段if user.user_name:user_to_update.name = user.user_name  # 修改用户的名称字段# 提交更改session.commit()return {"code": 200, "success": "User updated successfully"}except Exception as e:return {"code": 400, "error": f"User updated error{e}"}

调用FastAPI接口：

在这里插入图片描述

数据库修改结果：

在这里插入图片描述

十四.删除数据

代码如下

# main.py
# 5. 删除用户和密码
@app.delete("/delete_user/{user_id}")
def delete_user(user_id: str):db = orm.OrmManager()# 查询并删除用户和密码user = db.query_records(orm.MUH_User, filters={"id": user_id})password = db.query_records(orm.MUH_User_Psw, filters={"user_id": user_id})if not user or not password:raise HTTPException(status_code=404, detail="User not found")db.delete_records(user+password)  # 删除用户及其密码 +好为两个列表合并为一个列表return {"code": 200, "success": "User and password deleted successfully"}

调用FastAPI

在这里插入图片描述

数据库结果

在这里插入图片描述

十五.多表联合查询

代码如下：

# main.py
# 6.查询所有用户和密码
@app.get("/get_all_users_and_passwords")
def get_all_users_and_passwords():db = orm.OrmManager()# 联接查询，获取用户及其密码users_with_passwords = db.fetch_session().query(orm.MUH_User, orm.MUH_User_Psw).join(orm.MUH_User_Psw, orm.MUH_User.id == orm.MUH_User_Psw.user_id).all()if not users_with_passwords:raise HTTPException(status_code=404, detail="No users found")# 构建返回的数据result = []for user, password in users_with_passwords:result.append({"id": user.id,"user_name": user.user_name,"user_gender": user.user_gender,"user_age": user.user_age,"user_imgs": user.user_imgs,"user_psw": password.user_psw  })return {"code": 200, "success": "Users and passwords fetched successfully", "data": result}

调用FastAPI结果：

在这里插入图片描述

十六.完整的main.py代码

#main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from utils import orm_helper as orm
from datetime import datetime
import uuid
from sqlalchemy.orm import joinedload
from sqlalchemy import joinapp = FastAPI()# 1. 创建表
@app.get("/create_table")
def create_table():db = orm.OrmManager()db.create_table()return {"code": 200, "success": "Tables created successfully", "data": ""}# 2. 新增用户和密码
class UserCreate(BaseModel):user_name: struser_gender: struser_age: intuser_imgs: Optional[str] = Noneuser_psw: str@app.post("/create_user")
def create_user(user: UserCreate):db = orm.OrmManager()# 创建用户new_user = orm.MUH_User(id=str(uuid.uuid4()),user_name=user.user_name,user_gender=user.user_gender,      user_age=user.user_age,user_imgs=user.user_imgs)# 创建密码new_psw = orm.MUH_User_Psw(id=str(uuid.uuid4()),user_id=new_user.id,user_psw=user.user_psw)# 插入到数据库db.add_records([new_user, new_psw])return {"code": 200, "success": "User and password created successfully"}# 3. 查询用户信息
@app.get("/get_user/{user_id}")
def get_user(user_id: str):db = orm.OrmManager()# 查询用户和密码user = db.query_records(orm.MUH_User, filters={"id": user_id})password = db.query_records(orm.MUH_User_Psw, filters={"user_id": user_id})if user and password:return {"user": user[0], "password": password[0].user_psw}else:raise HTTPException(status_code=404, detail="User not found")# 4. 更新用户信息
class UserUpdate(BaseModel):id:Optional[str] = Noneuser_name: Optional[str] = None@app.put("/update_user/{user_id}")
def update_user(user: UserUpdate):try:db = orm.OrmManager()session = db.fetch_session()user_to_update  = session.query(orm.MUH_User).filter_by(id = user.id).first()print(user_to_update )user_to_update .user_name = user.user_nameif not user_to_update:raise HTTPException(status_code=404, detail="User not found")# 更新用户字段if user.user_name:user_to_update.name = user.user_name  # 修改用户的名称字段# 提交更改session.commit()return {"code": 200, "success": "User updated successfully"}except Exception as e:return {"code": 400, "error": f"User updated error{e}"}# 5. 删除用户和密码
@app.delete("/delete_user/{user_id}")
def delete_user(user_id: str):db = orm.OrmManager()# 查询并删除用户和密码user = db.query_records(orm.MUH_User, filters={"id": user_id})password = db.query_records(orm.MUH_User_Psw, filters={"user_id": user_id})if not user or not password:raise HTTPException(status_code=404, detail="User not found")db.delete_records(user+password)  # 删除用户及其密码 +好为两个列表合并为一个列表return {"code": 200, "success": "User and password deleted successfully"}# 6.查询所有用户和密码
@app.get("/get_all_users_and_passwords")
def get_all_users_and_passwords():db = orm.OrmManager()# 联接查询，获取用户及其密码users_with_passwords = db.fetch_session().query(orm.MUH_User, orm.MUH_User_Psw).join(orm.MUH_User_Psw, orm.MUH_User.id == orm.MUH_User_Psw.user_id).all()if not users_with_passwords:raise HTTPException(status_code=404, detail="No users found")# 构建返回的数据result = []for user, password in users_with_passwords:result.append({"id": user.id,"user_name": user.user_name,"user_gender": user.user_gender,"user_age": user.user_age,"user_imgs": user.user_imgs,"user_psw": password.user_psw  # 获取密码})return {"code": 200, "success": "Users and passwords fetched successfully", "data": result}

十七.总结

关于python ORM系列，从背景介绍，到数据库安装，到pymysql使用，到最终的ORM，一起写了好几个晚上，终于实现了ORM 让python操纵数据库像操作对象一样！看到这里的应该都是付费用户，请大家多多关注多多点赞 ！

如果各位大佬觉得文章还不错，能不能给个赏，让我动力澎湃 更新更快更好更精彩的内容给大家！

代码包 放到我的资源里面了，大家找找如果审核通过，应该可以下！

创作整理不易，请大家多多关注 多多点赞，有写的不对的地方欢迎大家补充，我来整理，再次感谢！

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

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

相关文章

从哈佛哲学系到蛋白质设计大师，David Baker：AlphaFold令我深刻认识到深度学习的力量

从哈佛哲学系到蛋白质设计大师，David Baker：AlphaFold令我深刻认识到深度学习的力量

要说谁是引领蛋白质设计的世界级大师&#xff0c;美国华盛顿大学的 David Baker 教授可谓是当之无愧&#xff0c;作为该领域的顶级专家&#xff0c;Baker 在蛋白质方向发表研究论文 700 余篇&#xff0c;引用量累计超 17.7 万。今年 10 月&#xff0c;因其在蛋白质设计方面的卓…
阅读更多...
【测试框架篇】单元测试框架pytest（2）：用例编写

【测试框架篇】单元测试框架pytest（2）：用例编写

一、 前言 前面一章我们介绍了pytest环境安装和配置&#xff0c;并在pycharm里面实现了我们第一个pytest脚本。但是有些童鞋可能在编写脚本的时候遇到了问题&#xff0c;本文会讲一下我们编写pytest用例时需要遵守哪些既定的规则&#xff0c;同时这个规则也是可以修改的。 二…
阅读更多...
实现LiDAR和多视角摄像头数据的对齐、可控X-DRIVE：用于驾驶场景的跨模态一致多传感器数据合成

实现LiDAR和多视角摄像头数据的对齐、可控X-DRIVE：用于驾驶场景的跨模态一致多传感器数据合成

Abstract 近年来&#xff0c;扩散模型在合成驾驶场景中的LiDAR点云或摄像头图像数据方面取得了进展。尽管这些模型在单一模态数据的边际分布建模方面取得成功&#xff0c;但对不同模态之间互相依赖关系的探索仍然不足&#xff0c;而这种依赖关系能够更好地描述复杂的驾驶场景。…
阅读更多...
稳恒磁场（1）

稳恒磁场（1）

物理概念 磁场是物质性的。 地磁场&#xff08;与地磁场正负极相反&#xff09;与磁偏角&#xff08;一般为0到11度&#xff09; 磁感应强度&#xff1a; 单位为特斯拉&#xff08;T&#xff09;&#xff0c;另一个常用单位是高斯&#xff08;G&#xff09;且1T 10^4 G 物…
阅读更多...
自动驾驶系列—自动驾驶中的短距离感知：超声波雷达的核心技术与场景应用

自动驾驶系列—自动驾驶中的短距离感知：超声波雷达的核心技术与场景应用

&#x1f31f;&#x1f31f; 欢迎来到我的技术小筑&#xff0c;一个专为技术探索者打造的交流空间。在这里&#xff0c;我们不仅分享代码的智慧&#xff0c;还探讨技术的深度与广度。无论您是资深开发者还是技术新手&#xff0c;这里都有一片属于您的天空。让我们在知识的海洋中…
阅读更多...
多语言爬取淘宝价格信息 python 比价api接入指南

多语言爬取淘宝价格信息 python 比价api接入指南

以下是爬取淘宝价格信息及接入淘宝比价 API 的一般步骤&#xff1a; 传统爬虫方式获取价格信息&#xff08;不建议大量使用&#xff0c;可能违反淘宝规定&#xff09;&#xff1a; 分析目标页面 URL&#xff1a;在淘宝搜索框输入关键词后&#xff0c;观察页面的 URL 结构。例如…
阅读更多...
Java List——针对实习面试

Java List——针对实习面试

目录 Java ListJava List的三种主要实现是什么&#xff1f;它们各自的特点是什么&#xff1f;Java List和Array&#xff08;数组&#xff09;的区别&#xff1f;Java List和Set有什么区别&#xff1f;ArrayList和Vector有什么区别&#xff1f;什么是LinkedList&#xff1f;它与…
阅读更多...
如何在Linux系统中安装微信

如何在Linux系统中安装微信

官方版微信的安装 好消息是&#xff0c;现在微信已经发布了官方的Linux版本&#xff0c;大家可以直接通过官方网站下载并安装&#xff0c;避免了以前繁琐的第三方工具安装步骤。 1.1 下载官方版微信 微信&#xff0c;是一个生活方式 选择Linux-> X86 1.2 安装微信 提前…
阅读更多...
java双向链表解析实现双向链表的创建含代码

java双向链表解析实现双向链表的创建含代码

双向链表 一.双向链表二.创建MyListCode类实现双向链表创建一.AddFirst创建&#xff08;头插法&#xff09;二.AddLast创建&#xff08;尾叉法&#xff09;三.size四.remove(指定任意节点的首位删除)五.removeAll(包含任意属性值的所有删除)六.AddIndex(给任意位置添加一个节点…
阅读更多...
hhdb数据库介绍(2-2)

hhdb数据库介绍(2-2)

数据高可用服务 HHDB Server在计算节点、数据节点、配置库等层次提供全面的高可用保障。提供完善的心跳检测、故障切换对存储节点同步追平判断、全局自增序列在故障时自动跳号、客户端连接Hold等机制&#xff0c;保障数据服务的可用性与数据的一致性。 计算节点服务高可用 H…
阅读更多...
精挑细选的100道软测高频面试题，面试前你肯定用得上

精挑细选的100道软测高频面试题，面试前你肯定用得上

测试技术面试题 1、什么是兼容性测试&#xff1f;兼容性测试侧重哪些方面&#xff1f; 2、我现在有个程序&#xff0c;发现在 Windows 上运行得很慢&#xff0c;怎么判别是程序存在问题还是软硬件系统存在问题&#xff1f; 3、测试的策略有哪些&#xff1f; 4、正交表测试用…
阅读更多...
STM32获取SHT3X温湿度芯片数据

STM32获取SHT3X温湿度芯片数据

目录 一、概述 二、单次数据采集模式的测量 1、配置说明 2、代码实现方式 三、周期性数据采集模式的测量 1、配置说明 2、代码实现方式 四、完整代码下载链接 一、概述 SHT3X是Sensirion公司推出的一款高精度、完全校准的温湿度传感器&#xff0c;基于CMOSens技术。它提…
阅读更多...
[原创]手把手教学之前端0基础到就业——day11（ Javascript ）

[原创]手把手教学之前端0基础到就业——day11（ Javascript ）

文章目录 day11(Javascript)01Javascript①Javascript是什么②JavaScript组成③ Javascript的书写位置1. 行内式 (不推荐)2 . 内部位置使用 ( 内嵌式 )3. 外部位置使用 ( 外链式 ) 02变量1. 什么是变量2. 定义变量及赋值3. 注意事项4. 命名规范 03输入和输出1) 输出形式12) 输出…
阅读更多...
[JAVAEE] 面试题(五) - HashMap, Hashtable, ConcurrentHashMap

[JAVAEE] 面试题(五) - HashMap, Hashtable, ConcurrentHashMap

目录 一. Hashtable1.1 Hashtable效率低下的原因: 二. ConcurrentHashMap2.1 ConcurrentHashMap更高效的原因: 三. HashMap, Hashtable, ConcurrentHashMap 之间的区别 HashMap是线程不安全的. 在多线程环境下, 使用: HashtableConcurrentHashMap 来确保线程安全. 一. Hashta…
阅读更多...
Vue 2 —Vue Router 页面导航和参数传递

Vue 2 —Vue Router 页面导航和参数传递

当从A页面跳转到B页面的时候把数据也一起传递过去&#xff0c;可用Vue Router 功能&#xff1a; 一、. this.$router.push 方法 Vue Router 是 Vue.js 的官方路由管理器&#xff0c;允许你在应用中进行页面导航&#xff08;即跳转到不同的 URL 路径&#xff09;。 this.$rout…
阅读更多...
Local Transfer 致力于更加便捷地共享传输文件

Local Transfer 致力于更加便捷地共享传输文件

软件主页&#xff1a;https://illusionna.github.io/LocalTransfer
阅读更多...
[AcWing算法基础课]动态规划之01背包

[AcWing算法基础课]动态规划之01背包

题目链接&#xff1a;01背包 有 N 件物品和一个容量是 V 的背包。每件物品只能使用一次。第 i 件物品的体积是 vi&#xff0c;价值是 wi。求解将哪些物品装入背包&#xff0c;可使这些物品的总体积不超过背包容量&#xff0c;且总价值最大。输出最大价值。 首先&#xff0c;我们…
阅读更多...
标准、高效的管理测试用例和活动

标准、高效的管理测试用例和活动

送您一份新人礼&#xff0c;自动化测试平台限时免费体验~ 本文主要介绍测试用例管理的基础知识和基本使用方法&#xff0c;帮助您快速管理测试用例及活动。 操作流程 用例管理的主要使用流程如下&#xff1a; 1.新建测试用例 2.评审测试用例 3.创建测试计划 4.执行测试计划 5…
阅读更多...
如何在jupyter notebook切换python环境

如何在jupyter notebook切换python环境

目录 1、切换到目标python环境&#xff0c;假设我的是叫“tf” C:\Users\hello>activate tf(tf) C:\Users\hello>2、安装notebook内核包 (tf) C:\Users\hello>pip install ipykernel3、将环境加入到notebook中 python -m ipykernel install --user --name pytorch --…
阅读更多...
windows工具 -- 使用SpaceSniffer查看哪些文件夹占用那么大空间, 再也不用右键属性了

windows工具 -- 使用SpaceSniffer查看哪些文件夹占用那么大空间, 再也不用右键属性了

目的 C盘不知道哪些文件夹占用了那么多空间, 右键属性扫描太慢了 效果 运行效果 静态截图 下载使用 下载 SpaceSniffer https://github.com/redtrillix/SpaceSniffer/releases 解压到文件夹后, 双击运行
阅读更多...
最新文章

Copyright @ 2022~2023