慕慕手记项目日记 熟悉SQLAIChemy ORM框架 2025-3-5
SQLAIChemy框架
SQLAlchemy 是一个功能强大的 Python SQL 工具包和对象关系映射（ORM）框架，用于简化数据库操作。它提供了全套企业级持久化模式，既支持纯 SQL 开发，也支持高级 ORM 操作，是 Python 生态中最流行的数据库工具之一。
安装SQLAlChemy
使用以下命令进行安装
pip install sqlalchemy
安装postman
如果是win11可以使用以下命令进行安装：
winget install post
也可以选择打开官网：
Download Postman | Get Started for Free
进去点击下载即可。后续还需要创建账号登录，此部分省略。
开发用户登录功能
这个案例展示了如何使用sqlalchemy来创建基类和动态映射User表，还有创建会话对象和使用 scoped_session 确保线程安全的会话管理。最后通过postman发送测试数据，这个客户端收到json请求后会立马请求我们的mysql数据库。由于暂时使用明文存储，没有使用md5进行加密。所以省去了md5函数验证的步骤
import json
from flask import Flask, request, jsonify
import sqlalchemy
from sqlalchemy import Table
from sqlalchemy.engine import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session, declarative_base# 创建 SQLAlchemy 基类，用于映射表结构
Base = declarative_base()# 创建 Flask 应用实例
app = Flask(__name__)# 创建数据库连接
engine = create_engine("mysql+pymysql://root:123456@172.27.13.88:3306/mumushouji", echo=True)# 创建会话工厂
session = sessionmaker(engine)# 使用 scoped_session 确保线程安全的会话管理
db_session = scoped_session(session)# 定义 User 类，映射到数据库中的 USER 表
class User(Base):__table__ = Table("USER", Base.metadata, autoload_with=engine) # 自动加载表结构@app.route("/login", methods=["POST"]) # 建议仅保留 POST 方法
def login():"""处理用户登录请求。请求方法：POST请求体：JSON 格式，包含 "username" 和 "password" 字段返回值：JSON 格式的响应，包含成功或错误信息"""try:# 获取并解析请求体中的 JSON 数据request_data = request.get_json()if not request_data:return jsonify({"error": "Invalid or empty JSON"}), 400 # 如果 JSON 数据无效或为空，返回 400 错误# 检查必填字段是否存在于请求数据中required_fields = ["username", "password"]for field in required_fields:if field not in request_data:return jsonify({"error": f"Missing field: {field}"}), 400 # 缺少必填字段，返回 400 错误# 提取用户名和密码user_input_name = request_data["username"]input_password = request_data["password"]print(user_input_name, input_password) # 打印用户名和密码（仅用于调试）# 查询用户是否存在result = db_session.query(User).filter_by(username=user_input_name).first()if not result:return jsonify({"error": "User not found"}), 404 # 用户不存在，返回 404 错误# 验证密码（假设密码是明文存储，实际应使用哈希）if result.password != input_password:return jsonify({"error": "Invalid password"}), 401 # 密码不匹配，返回 401 错误# 登录成功，返回成功消息return jsonify({"message": "登录成功"})except Exception as e:# 捕获所有异常，并返回服务器内部错误return jsonify({"error": f"Internal server error: {str(e)}"}), 500if __name__ == '__main__':app.run(debug=True) # 启动 Flask 应用，启用调试模式方便开发数据插入
这里是一个插入数据的逻辑，先拿到请求数据里面的内容，然后单独对密码进行md5加密，这次不再对密码做明文存储。然后再将数据依次放进User类里面，然后将数据添加到数据库，最后提交。
@app.route("/register", methods=["POST"])
def register():request_data = request.datarequest_data = json.loads(request_data)print(request_data)#设置用户名和密码username = request_data["username"]password = request_data["password"]nickname = request_data["nickname"]picture = request_data["picture"]#暂时只使用四个参数，后面可以添加更多参数password = hashlib.md5(password.encode()).hexdigest()insert_param = {"username": username,"password": password,"nickname": nickname,"picture": picture}user = User(**insert_param)db_session.add(user)db_session.commit()return "注册成功"
可以看到，数据已经插入进来了。
关于User(**insert_param)的含义，以下是AI的解释：
在Python中，User(**insert_param) 这种语法使用了可变参数传递的方式（也称为解包操作符）。这里是如何工作的：
假设你有一个类 User，它可能定义了一些初始化参数，比如这样：
class User:def __init__(self, username, password, nickname, picture):self.username = usernameself.password = passwordself.nickname = nicknameself.picture = picture
现在，如果你想创建一个 User 类的实例，并且你有一个字典 insert_param 包含了所有必要的键值对，你可以使用解包操作符 ** 来将这个字典解包为关键字参数传递给 User 类的构造函数。例如：
insert_param = {"username": "user12345","password": "pass$W0rd6789","nickname": "CoolUser_123","picture": "https://example.com/images/user_profile_12345.png"
}user = User(**insert_param)
这里的 **insert_param 会将字典中的键值对解包成关键字参数的形式，相当于这样调用：
user = User(username="user12345", password="pass$W0rd6789", nickname="CoolUser_123", picture="https://example.com/images/user_profile_12345.png")
这种技术非常有用，尤其是在处理动态生成的参数或者当你想要从字典中提取数据来创建对象实例时。它提供了一种简洁的方式来传递一组名称-值对作为关键字参数给函数或方法。确保字典中的键名与类构造函数期待的关键字参数名称相匹配是很重要的，否则将会导致错误，比如 TypeError 提示找不到正确的参数。
删除和修改数据
这两个例子展示了如何通过SQLAlchemy简单高效地执行删除和更新操作。重要的是要记得在完成这些操作后调用db_session.commit()来提交更改，否则更改不会被保存到数据库中。此外，在生产环境中执行删除或更新操作时应格外小心，最好先备份数据，确保操作的安全性和准确性。
#删除数据
del_data = db_session.query(User).filter_by(username="dazhoulaos222").delete()
db_session.commit()
print("删除成功")#修改数据
change_data = db_session.query(User).filter_by(username="test_user").update({"nickname": "test_user_change"})
db_session.commit()
print("修改成功")
单表查询操作补充
from sqlalchemy import and_# 查询操作
def query_users():# 查询 User 表中的所有记录users = db_session.query(User).all()for user in users:print(f"User ID: {user.user_id}, Username: {user.username}")# 根据条件查询，例如 username 等于 'test_user'user = db_session.query(User).filter_by(username='test_user').first()if user:print(f"Found user: {user.username}, Nickname: {user.nickname}")# 统计 user_id 小于 10 的用户数量count = db_session.query(User).filter(User.user_id < 10).count()print(f"Number of users with user_id < 10: {count}")# 插入操作
def insert_user():new_user = User(username='new_username', password='secure_password', nickname='Newbie', picture='http://example.com/pic.png')db_session.add(new_user)db_session.commit()print("Inserted new user.")# 更新操作
def update_user():# 使用 filter_by 进行更新user = db_session.query(User).filter_by(username='user_to_update').first()if user:user.nickname = 'UpdatedNickname'db_session.commit()print("User updated.")# 或者直接使用 update 方法批量更新db_session.query(User).filter_by(username='another_user_to_update').update({"nickname": "AnotherUpdatedNickname"})db_session.commit()print("Users batch updated.")# 删除操作
def delete_user():# 删除单个实体user = db_session.query(User).filter_by(username='username_to_delete').first()if user:db_session.delete(user)db_session.commit()print("User deleted.")# 批量删除db_session.query(User).filter(User.username.like('delete_me_%')).delete()db_session.commit()print("Batch deletion completed.")if __name__ == "__main__":try:# 执行查询query_users()# 执行插入insert_user()# 执行更新update_user()# 执行删除delete_user()except Exception as e:db_session.rollback() # 如果发生错误则回滚print(f"An error occurred: {e}")finally:db_session.close() # 确保关闭数据库会话
