【FastAPI】使用 FastAPI 和 SQLAlchemy 记录数据库操作日志：基于装饰器的实现

🕗 发布于 2024-09-19 12:54 数据库 fastapi oracle

在后端开发中，记录用户对数据库的操作日志（如增、删、改）不仅是调试和监控的有效工具，还能提升系统的可维护性和安全性。本文将介绍如何在 FastAPI 中使用装饰器来记录数据库操作日志，并保留修改前和修改后的内容。

一、为什么需要操作日志？

审计：操作日志能够追踪谁在何时对数据进行了哪些修改，尤其在处理敏感数据时非常有必要。
调试：当系统出现问题时，通过日志可以快速查明问题根源，尤其是在涉及数据一致性的问题时。
安全：可以识别恶意操作或非授权修改，从而采取相应的安全措施。

通过记录操作前后的数据状态，我们能够清楚地知道数据库的变化，并且可以在系统出错或需要回滚时进行详细分析。

二、项目设置

1. 数据库设置

首先，我们创建一个日志记录表，用于保存用户操作的详细信息。我们将使用 PostgreSQL 来存储数据（当然，你可以使用任何你熟悉的数据库系统）。

CREATE TABLE operation_logs (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL, -- 操作用户的ID
    operation_type VARCHAR(50) NOT NULL, -- 操作类型，INSERT, UPDATE, DELETE
    table_name VARCHAR(100) NOT NULL, -- 操作的表名
    record_id INTEGER NOT NULL, -- 被操作记录的ID
    operation_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- 操作时间
    before_change TEXT, -- 修改前的数据
    after_change TEXT -- 修改后的数据
);

2. 使用 SQLAlchemy 进行数据库操作

为了更好地与数据库交互，我们使用 SQLAlchemy 作为 ORM。这里是基本的 FastAPI 与 SQLAlchemy 的配置：

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import sessionmaker

app = FastAPI()

# 数据库连接配置
DATABASE_URL = "postgresql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. 编写日志记录装饰器

装饰器是一种非常简洁且优雅的方式，可以将日志记录与业务逻辑分离。接下来，我们将创建一个装饰器，专门用于记录数据库增删改操作时的日志信息。

from functools import wraps
from datetime import datetime
import json

def log_changes(operation_type):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            db = kwargs.get('db')  # 获取数据库会话
            user_id = kwargs.get('user_id')  # 获取用户ID
            table_name = kwargs.get('table_name')  # 操作的表名
            record_id = kwargs.get('record_id')  # 操作记录的ID
            
            # 获取修改前的数据
            before_change = func.__name__ == "delete_item" and "N/A" or db.query(Table(table_name)).filter_by(id=record_id).first()
            before_change_data = json.dumps(dict(before_change)) if before_change else "N/A"

            result = func(*args, **kwargs)  # 执行实际的增删改操作

            # 获取修改后的数据
            after_change = func.__name__ == "delete_item" and "N/A" or db.query(Table(table_name)).filter_by(id=record_id).first()
            after_change_data = json.dumps(dict(after_change)) if after_change else "N/A"
            
            # 记录操作日志
            log_operation(db, user_id, operation_type, table_name, record_id, before_change_data, after_change_data)

            return result
        return wrapper
    return decorator

4. 日志记录函数

我们还需要编写一个 log_operation 函数，用来将日志信息插入到数据库中。

def log_operation(db, user_id, operation_type, table_name, record_id, before_change, after_change):
    query = f"""
    INSERT INTO operation_logs (user_id, operation_type, table_name, record_id, operation_time, before_change, after_change)
    VALUES ({user_id}, '{operation_type}', '{table_name}', {record_id}, '{datetime.now()}', '{before_change}', '{after_change}');
    """
    db.execute(query)
    db.commit()

5. 应用装饰器

现在我们可以将装饰器应用到 FastAPI 的路由中，以便在用户执行增删改操作时自动记录操作日志。

@app.post("/items/")
@log_changes("INSERT")
def create_item(item: dict, user_id: int, db: SessionLocal = Depends(get_db)):
    # 执行插入操作
    table_name = "items"
    record_id = 123  # 插入记录后生成的ID

    # 实际插入代码 (示例)
    db.execute(f"INSERT INTO {table_name} (name, value) VALUES ('{item['name']}', '{item['value']}')")
    db.commit()

    return {"status": "success", "record_id": record_id}

@app.put("/items/{item_id}")
@log_changes("UPDATE")
def update_item(item_id: int, updated_item: dict, user_id: int, db: SessionLocal = Depends(get_db)):
    # 执行更新操作
    table_name = "items"
    
    # 实际更新代码 (示例)
    db.execute(f"UPDATE {table_name} SET name = '{updated_item['name']}', value = '{updated_item['value']}' WHERE id = {item_id}")
    db.commit()

    return {"status": "updated", "item_id": item_id}

@app.delete("/items/{item_id}")
@log_changes("DELETE")
def delete_item(item_id: int, user_id: int, db: SessionLocal = Depends(get_db)):
    # 执行删除操作
    table_name = "items"
    
    # 实际删除代码 (示例)
    db.execute(f"DELETE FROM {table_name} WHERE id = {item_id}")
    db.commit()

    return {"status": "deleted", "item_id": item_id}

三、总结

通过本文的示例，我们展示了如何在 FastAPI 中结合 SQLAlchemy 使用装饰器来记录数据库的增删改操作日志。这种方法的优势在于它与业务逻辑分离，使用简单且易于扩展。此外，日志表中保存了修改前后的数据，使得系统在调试和审计时更加高效和透明。

希望这篇文章能帮助你在项目中实现类似的功能！如果你有任何问题或建议，欢迎在评论区留言。

原文地址：https://blog.csdn.net/h1773655323/article/details/142263006

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：微服务保护学习笔记(三)流控效果(warm up、排队等待)、热点参数限流
下一篇：Datawhale X 南瓜书 task01学习笔记

leetcode289:生命游戏
根据，简称为，是英国数学家约翰·何顿·康威在 1970 年发明的细胞自动机。给定一个包含m × n个格子的面板，每一个格子都可以看成是一个细胞。每个细胞都具有一个初始状态：1即为（live），或0即为
阅读更多2024-10-20
MongoDB数据恢复
注意：两个MongoDB的版本要一致，本文使用的是mongo:4.2.24。先把K8S上面的MongoDB 容器停止（可以把副本改成0）。1、将容器挂载MongoDB的数据目录备份到本地。经常是数据文
阅读更多2024-10-20
C#中实现事务
C#中实现事务
阅读更多2024-10-20
【LeetCode每日一题】——560.和为 K 的子数组
给你一个整数数组 nums 和一个整数 k ，请你统计并返回该数组中和为 k 的子数组的个数。子数组是数组中元素的连续非空序列。
阅读更多2024-10-20
「漏洞复现」满客宝智慧食堂系统 selectUserByOrgId 未授权访问漏洞
请勿利用文章内的相关技术从事非法测试，由于传播、利用此文所提供的信息而造成的任何直接或者间接的后果及损失，均由使用者本人负责，作者不为此承担任何责任。工具来自网络，安全性自测，如有侵权请联系删除。本次
阅读更多2024-10-20
React面试题目（从基本到高级）
React前端面试常见题目涵盖了React的基础概念、组件、状态管理、生命周期、性能优化等多个方面。
阅读更多2024-10-20
12.个人博客系统（Java项目基于spring和vue）
1 在校学习的学生，可用于日常学习使用或是毕业设计使用 2 毕业一到两年的开发人员，用于锻炼自己的独立功能模块设计能力，增强代码编写能力。 3 亦可以部署为商化项目使用。 4 需要完整资料及源码
阅读更多2024-10-20
YoloV8改进策略：注意力改进|DeBiFormer，可变形双级路由注意力|引入DeBiLevelRoutingAttention注意力模块（全网首发）
本次改进的核心在于将DeBiLevelRoutingAttention模块嵌入到YoloV8的主干网络中，具体位置是在SPPF（Spatial Pyramid Pooling Fast）模块之后。这一
阅读更多2024-10-20
word取消自动单词首字母大写
情况说明：在word输入单词后首字母会自动变成大写取消单词首字母大写步骤：（1）点击菜单栏文件（2）点击“更多”——>“选项”（3）点击“校对”——>“自动更正选项”（4）取消“句首字母大
阅读更多2024-10-20
web前端网页用户注册页面
【代码】web前端网页用户注册页面。
阅读更多2024-10-20