FastAPI+Sqlalchemy执行【Mysql】原生sql

一. 前言

当有的复杂sql 用orm不好写出来的时候，此时想要用原生sql查询
- 原生sql查询，查出的结果是对象
- 原生sql查询，查询结果列表套元组

当你使用 SQLAlchemy 并希望通过创建的 session 对象来执行原生 SQL 语句时，可以使用 session.execute() 方法。这种方式允许你直接在数据库中运行 SQL 语句，同时仍然能够享受 SQLAlchemy 会话管理和事务控制的好处。以下是如何使用 session.execute() 方法执行查询、插入、更新和删除操作的示例。

二. 使用

from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engineengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/db001", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

2.1. 执行原生sql方式一：

2.1.1 查询方式一：

cursor = session.execute('select * from users')
result = cursor.fetchall()
print(result) #列表套元组
# 添加
cursor = session.execute('insert into users(name,email) values(:name,:email)',params={"name": 'lqz', 'email': '3333@qq.com'})
session.commit()
print(cursor.lastrowid)

2.1.2. 基本使用方式

1.执行查询操作

def execute_query(sql: str):try:result = session.execute(sql)# 对于查询操作，通过fetchall()获取所有结果rows = result.fetchall()# 将结果转换为字典列表results = [dict(row) for row in rows]return resultsexcept Exception as e:session.rollback()raisefinally:session.close()# 调用函数执行查询
sql_query = "SELECT * FROM your_table_name"
query_results = execute_query(sql_query)
print(query_results)

2.执行插入操作

def execute_insert(sql: str, values: dict):try:session.execute(sql, values)session.commit()except Exception as e:session.rollback()raisefinally:session.close()# 插入数据示例
sql_insert = "INSERT INTO your_table_name (column1, column2) VALUES (:value1, :value2)"
values = {"value1": "data1", "value2": "data2"}
execute_insert(sql_insert, values)

3.执行更新操作

def execute_update(sql: str, values: dict):try:session.execute(sql, values)session.commit()except Exception as e:session.rollback()raisefinally:session.close()# 更新数据示例
sql_update = "UPDATE your_table_name SET column1 = :value1 WHERE column2 = :value2"
values = {"value1": "new_data1", "value2": "data2"}
execute_update(sql_update, values)

4.执行删除操作

def execute_delete(sql: str, values: dict):try:session.execute(sql, values)session.commit()except Exception as e:session.rollback()raisefinally:session.close()# 使用示例
sql_delete = "DELETE FROM your_table_name WHERE column1 = :value"
values = {"value": "specific_value_to_delete"}
execute_delete(sql_delete, values)