You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Flask REST API能否采用文本文件作为客户端授权验证的数据库?

最优方案:用JSON文件作为轻量存储+自动同步机制

你的需求完全适合用JSON文件作为存储方案,既避免了修改代码重启服务器的麻烦,又比常规数据库更轻量。下面是具体实现思路和代码示例:

一、基础JSON文件读写实现

首先把用户数据从代码里移到单独的users.json文件,格式和你原来的字典一致:

{
  "user_name0": ["rand_pass_0", "hashed_pass_0", "2025-12-31"],
  "user_name1": ["rand_pass_1", "hashed_pass_1", "2024-06-30"]
}

然后写两个工具函数负责加载和保存数据,同时加锁避免并发读写冲突:

import json
import os
from threading import Lock

# 全局变量:用户字典和文件锁
user_db = {}
db_lock = Lock()
db_file_path = "users.json"
last_modified_time = 0

def load_user_db():
    global user_db, last_modified_time
    with db_lock:
        if not os.path.exists(db_file_path):
            # 初始化空文件
            save_user_db({})
        with open(db_file_path, "r") as f:
            user_db = json.load(f)
        last_modified_time = os.path.getmtime(db_file_path)

def save_user_db(data):
    with db_lock:
        with open(db_file_path, "w") as f:
            json.dump(data, f, indent=2)

在Flask启动时自动加载数据:

from flask import Flask

app = Flask(__name__)

# 启动时加载用户数据
load_user_db()

二、自动检测文件变更并同步

有两种简单的实现方式,按需选择:

方式1:每次授权请求前检查(适合请求量不大的场景)

在处理授权的接口里,先检查文件是否被修改,是则重新加载:

@app.route("/auth", methods=["POST"])
def auth():
    # 先检查用户数据库文件是否更新
    current_mtime = os.path.getmtime(db_file_path)
    if current_mtime != last_modified_time:
        load_user_db()
    
    # 后续授权逻辑:从user_db里取数据验证
    # ...
    return {"status": "success"}

方式2:后台线程定时检查(适合请求量较大或需要实时同步的场景)

用后台线程每隔一段时间(比如30秒)检查一次文件修改时间,自动更新:

import threading
import time

def watch_db_file():
    while True:
        time.sleep(30)  # 每30秒检查一次
        current_mtime = os.path.getmtime(db_file_path)
        if current_mtime != last_modified_time:
            load_user_db()

# 启动后台监控线程(设为守护线程,Flask退出时自动结束)
watch_thread = threading.Thread(target=watch_db_file, daemon=True)
watch_thread.start()

三、额外优化建议

  • 令牌密钥管理:把生成令牌的内部密码从代码里移到.env文件,用python-dotenv加载,更安全:
    # .env文件内容
    JWT_SECRET_KEY=your_secret_key_here
    
    # Flask代码里加载
    from dotenv import load_dotenv
    load_dotenv()
    SECRET_KEY = os.getenv("JWT_SECRET_KEY")
    
  • 异常处理:在load_user_db里加异常捕获,避免手动编辑JSON格式错误导致服务器崩溃:
    def load_user_db():
        global user_db, last_modified_time
        with db_lock:
            try:
                with open(db_file_path, "r") as f:
                    user_db = json.load(f)
                last_modified_time = os.path.getmtime(db_file_path)
            except json.JSONDecodeError:
                # 格式错误时回退到上次正确数据或初始化空字典
                app.logger.error("JSON文件格式错误,加载失败")
    

方案适配性总结

这个方案完全匹配你的需求:

  • 不用重启服务器就能通过手动编辑JSON文件新增/删除用户
  • 客户端请求始终访问内存中的user_db字典,性能不受影响
  • 比常规数据库轻量得多,没有多余的复杂度

内容的提问来源于stack exchange,提问作者Daniel Casasampera

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.20 23:54:32