查看: 110|回复: 1

FastAPI+Vue3全栈SQLite管理工具:JWT认证与只读SQL执行

[复制链接]
发表于 2 小时前 | 显示全部楼层 |阅读模式
在日常开发、测试和数据分析中,快速查看SQLite数据库的表结构、执行查询并保存常用SQL模板是常见需求。本文实现一个可运行的Python全栈Web应用:后端基于FastAPI提供RESTful API,使用SQLAlchemy定义数据模型,通过Passlib bcrypt加盐哈希存储密码,JWT Bearer Token实现接口鉴权;前端使用Vue 3 + Vite构建登录注册、数据库连接管理、SQL查询台等页面。整个项目前后端分离,所有查询严格限制为只读(仅允许SELECT、PRAGMA、EXPLAIN语句),避免误操作风险。

一、技术栈与项目结构
后端:FastAPI、SQLAlchemy、Pydantic、Uvicorn、Passlib、PyJWT
数据库:SQLite(系统业务库db_manager.db + 示例库demo_data.db)
前端:Vue 3、Vite、Fetch API
项目采用前后端分离架构,通过HTTP + JSON + Bearer Token通信。后端模块划分清晰:用户认证、连接管理、SQL模板、查询执行、审计统计。

二、数据库与数据模型设计
系统使用SQLAlchemy ORM定义以下核心表:
- users:用户表(id, username, email, hashed_password, created_at)
- db_connections:数据库连接表(id, name, db_type, database_path, owner_id)
- saved_queries:SQL模板表(id, title, sql_text, connection_id, owner_id)
- audit_logs:审计日志表(id, action, detail, owner_id, created_at)

其中db_connections和saved_queries通过owner_id关联到用户,确保数据隔离。模型文件backend/app/models.py中User定义如下:
  1. class User(Base):
  2.     __tablename__ = "users"
  3.     id = Column(Integer, primary_key=True, index=True)
  4.     username = Column(String(50), unique=True, index=True, nullable=False)
  5.     email = Column(String(120), unique=True, index=True, nullable=False)
  6.     hashed_password = Column(String(255), nullable=False)
  7.     created_at = Column(DateTime, default=datetime.utcnow)
复制代码

三、后端核心代码实现
1. 数据库连接配置
backend/app/database.py中使用SQLAlchemy创建引擎,并为SQLite特别设置check_same_thread=False以支持多线程:
  1. SQLALCHEMY_DATABASE_URL = "sqlite:///./db_manager.db"
  2. engine = create_engine(
  3.     SQLALCHEMY_DATABASE_URL,
  4.     connect_args={"check_same_thread": False},
  5. )
  6. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
  7. Base = declarative_base()
复制代码
同时定义get_db()作为FastAPI依赖注入,负责每次请求后关闭会话。

2. 密码哈希与JWT签发
在backend/app/crud.py中,使用Passlib的bcrypt方案进行密码哈希,安全性高且与标准JWT结合:
  1. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  2. def get_password_hash(password: str) -> str:
  3.     return pwd_context.hash(password)
  4. def verify_password(plain_password: str, hashed_password: str) -> bool:
  5.     return pwd_context.verify(plain_password, hashed_password)
复制代码
登录验证通过后,调用create_access_token生成JWT令牌,设置过期时间(通常30分钟,可通过ACCESS_TOKEN_EXPIRE_MINUTES配置):
  1. def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
  2.     to_encode = data.copy()
  3.     expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
  4.     to_encode.update({"exp": expire})
  5.     return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
复制代码

3. 接口鉴权实现
后端通过OAuth2PasswordBearer从请求中提取Bearer Token,解析后获取当前用户。main.py中定义了get_current_user依赖:
  1. def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
  2.     username = crud.decode_token(token)
  3.     if not username:
  4.         raise HTTPException(status_code=401, detail="无效或过期的登录凭证")
  5.     user = crud.get_user_by_username(db, username)
  6.     if not user:
  7.         raise HTTPException(status_code=401, detail="用户不存在")
  8.     return user
复制代码
只要在接口参数中加入current_user: models.User = Depends(get_current_user),该接口即受JWT保护。

4. 只读SQL执行安全控制
为了演示系统的安全性,后端对执行的SQL进行严格校验:仅允许SELECT、PRAGMA、EXPLAIN开头,否则抛出异常。代码位于查询执行模块:
  1. sql = request.sql_text.strip()
  2. first_word = sql.split(None, 1)[0].lower() if sql else ""
  3. if first_word not in {"select", "pragma", "explain"}:
  4.     raise ValueError("演示系统仅允许 SELECT/PRAGMA/EXPLAIN 只读语句")
复制代码
执行后通过sqlite3.Row转换为字典列表,便于前端动态渲染表格。

四、前端登录状态处理
前端采用Vue 3 + Vite,令牌统一保存在localStorage中,每次请求通过Fetch API自动添加Authorization头。核心封装在frontend/src/api.js:
  1. export function getToken() {
  2.     return localStorage.getItem('db_manager_token')
  3. }
  4. async function request(path, options = {}) {
  5.     const headers = options.headers || {}
  6.     const token = getToken()
  7.     if (token) headers.Authorization = `Bearer ${token}`
  8.     const response = await fetch(`${API_BASE}${path}`, { ...options, headers })
  9.     const data = await response.json().catch(() => ({}))
  10.     if (!response.ok) throw new Error(data.detail || '请求失败')
  11.     return data
  12. }
复制代码
App.vue在页面加载时检查localStorage中是否存在token,若存在则自动进入系统;退出登录时清除token并跳转到登录页。

五、部署与运行
1. 启动后端(Python 3.8+,建议使用虚拟环境):
  1. cd project/backend
  2. python3 -m venv .venv
  3. source .venv/bin/activate
  4. pip install -r requirements.txt
  5. uvicorn app.main:app --reload --host 127.0.0.1 --port 8000
复制代码
后端自动创建db_manager.db和demo_data.db,接口文档地址:http://127.0.0.1:8000/docs

2. 启动前端(需要Node.js):
  1. cd project/frontend
  2. npm install
  3. npm run dev
复制代码
前端开发服务器默认运行在http://127.0.0.1:5173

3. 使用示例
打开前端后注册账号,登录后在“数据库连接”表单中创建一个指向./demo_data.db的连接。然后在SQL查询台输入:SELECT * FROM employees;,点击“执行只读查询”即可看到示例员工数据。若需复用,可点击“保存为模板”,将SQL存入个人模板列表。

六、总结与扩展思路
本文基于FastAPI和Vue 3实现了一个功能完整的SQLite管理工具,覆盖了用户注册登录、JWT鉴权、只读查询执行、模板管理等核心需求。对于学习Python全栈开发的同学,项目中涉及的密码哈希、JWT、SQLAlchemy模型、FastAPI依赖注入、前端状态管理等都是关键知识点。后续可扩展支持多数据库类型、表结构浏览器、分页查询、角色权限控制、审计日志检索等功能,进一步增强实用性。
回复

使用道具 举报

发表于 2 小时前 | 显示全部楼层

Re: FastAPI+Vue3全栈SQLite管理工具:JWT认证与只读SQL执行

看了你的分享,这个全栈工具设计得很完整,从认证鉴权到只读查询的约束都考虑得很周到。尤其用SQLAlchemy+Passlib+JWT这套组合在FastAPI里很成熟,前端Vue3配合Fetch也简洁高效。 几个小问题想请教: 1. 你在 `crud.decode_token` 里是怎么处理JWT过期和签名校验的?直接解码还是用了`jwt.decode`的`algorithms`参数? 2. 前端Vue3部分是怎么实现Token持久化的?是存在localStorage还是用了Pinia配合拦截器? 3. 对于大量数据查询,有没有考虑过前端分页或后端流式返回? 另外,建议可以在README里加个快速启动脚本,这样新手克隆下来直接跑`uvicorn`和`npm run dev`就能用上。整体来说是个很实用的工具,收藏了!
回复 支持 反对

使用道具 举报

您需要登录后才可以回帖 登录 | 注册

本版积分规则

指导单位

江苏省公安厅

江苏省通信管理局

浙江省台州刑侦支队

DEFCON GROUP 86025

Hacking Group 021A

旗下站点

态势感知中心

应急响应中心

红盟安全

联系我们

官方QQ群:112851260

官方邮箱:security#ihonker.org(#改成@)

官方核心成员

关注微信公众号

Archiver|手机版|小黑屋| ( 沪ICP备2021026908号 )

GMT+8, 2026-6-10 12:04 , Processed in 0.028826 second(s), 18 queries , Gzip On, Redis On.

Powered by ihonker.com

Copyright © 2015-现在.

  • 返回顶部