简介基于FastAPI框架的在线课程学习系统是一份完整的Python项目源码与说明文档适合有一定Python基础、正在做毕设或课程设计的开发者参考也可作为在线教育平台快速原型的搭建模板。压缩包共56个文件以17个py源码文件与32个pyd编译模块为主另含依赖清单、项目说明及配置文件整体仅4.37MB目录按功能模块清晰划分便于快速定位用户、课程、学习、互动、管理等核心业务代码。系统覆盖用户注册登录、课程增删改查、学习进度跟踪、笔记作业、讨论评价和管理员后台等完整业务链能够直观看到FastAPI如何组织路由、连接数据库、封装通用工具以及处理鉴权与日志等关键环节。目前已有35人学习通过研读源码和项目说明可以掌握JWT鉴权、日志配置、分层模型等实战技巧对提升FastAPI项目开发能力有切实帮助。1. 先把 FastAPI 在线课程学习系统的源码盘顺一个毕设级别的 FastAPI 项目最容易翻车的地方不在业务代码而在目录分层和 token 处理。拆开这个 zip 压缩包你会发现main.py只负责挂载路由业务被拆到routers、common、models三个目录正是这层拆法决定了你改一个功能要动几个文件。这份基于 FastAPI 的在线课程学习 python 源码把用户注册登录、课程增删改查、学习进度和笔记都塞进了不到十个核心文件里没有分布式组件也没有消息队列核心就是 FastAPI SQLAlchemy JWT。项目适合有 Python 基础、正在做课程设计或想快速理解 FastAPI 工程化套路的开发者。下面直接从工程布局、认证链路、课程 CRUD、环境配置一路讲到 Uvicorn 启动和接口验证中间会给出能直接抄的代码和参数说明。2. 用户认证JWT 工具、注册登录与依赖注入在线课程系统里用户模块是第一个要面对的硬骨头。课程可以匿名浏览但提交笔记、记录进度、发布评价都必须知道“你是谁”。这个项目把认证抽到了common/jwtTool.py再用 FastAPI 的依赖注入把解析出来的用户信息塞进路由函数这一层设计一旦理顺后面所有业务接口都会很省事。2.1 jwtTool.pytoken 生成与解析的核心common/jwtTool.py几乎会被所有业务模块引用它负责两件事签发 token、校验 token。用 PyJWT 实现时常见做法是直接封装两个函数import time import jwt SECRET_KEY your-secret-key # 生产环境必须从 config.py 读取 ALGORITHM HS256 EXPIRE_SECONDS 60 * 60 * 24 # 默认 24 小时过期 def create_token(user_id: int, role: str user) - str: payload { user_id: user_id, role: role, exp: int(time.time()) EXPIRE_SECONDS } return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM) def parse_token(token: str) - dict: try: return jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) except jwt.ExpiredSignatureError as e: raise ValueError(token 已过期) from e except jwt.InvalidTokenError as e: raise ValueError(无效 token) from e这段代码的要点可以从三个参数看exp是标准过期声明PyJWT 会在解码时自动校验不需要手动比较时间ALGORITHM常见用 HS256同一个密钥负责签名和验证适合单体后端SECRET_KEY如果写死在代码里日后很难轮换这个项目在config.py里统一管理jwtTool.py应当只保持引用关系。实际运行时FastAPI 路由不会直接调parse_token而是把它包进一个依赖函数。比如在routers/user.py里会有类似下面的代码用Depends把解析结果注入到视图函数中from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security HTTPBearer() def get_current_user(credentials: HTTPAuthorizationCredentials Depends(security)): payload parse_token(credentials.credentials) return {user_id: payload[user_id], role: payload[role]}这里HTTPBearer会从请求头Authorization: Bearer token中自动取证书省去自己写 header 解析的工作。如果Authorization缺失FastAPI 直接返回 403路由函数里连异常判断都不需要写。2.2 注册接口密码校验与入库用户注册的核心不是“往表里插一条记录”而是先解决两个问题密码不能明文存用户名不能重复。密码处理一般用passlib或 Python 自带的hashlib。这个项目以轻量为主所以你会看到类似pbkdf2_sha256的哈希方式from passlib.context import CryptContext from fastapi import APIRouter, HTTPException from schemas import UserCreate pwd_context CryptContext(schemes[pbkdf2_sha256], deprecatedauto) router APIRouter(prefix/user, tags[user]) router.post(/register) def register(user: UserCreate): if get_user_by_username(user.username): raise HTTPException(status_code400, detail用户名已被注册) hashed_password pwd_context.hash(user.password) create_user(usernameuser.username, hashed_passwordhashed_password) return {msg: 注册成功}UserCreate是 Pydantic 模型至少要校验username和password字段非空。注意这里必须返回明确的中文错误提示否则前端很难把“409/400”翻译成对用户友好的文案。get_user_by_username和create_user在models/crud.py中实现实际上就是 SQLAlchemy 的session.query与add/commit。这里有个经常被忽略的细节注册接口不要返回敏感字段。干脆用 Pydantic 的响应模型只暴露id、username、created_at。否则一旦后来加了hashed_password字段会直接泄露哈希。models/schemas.py里一般会区分UserCreate、UserOut两种模型这就是 FastAPI 推荐的项目结构。2.3 登录与依赖注入获取当前用户登录接口和注册不同它不做“创建”而是根据用户名查出用户、校验密码、签发 token。一个能直接用的写法如下router.post(/login) def login(form: UserLogin, session: Session Depends(get_db)): user get_user_by_username(session, form.username) if not user or not pwd_context.verify(form.password, user.hashed_password): raise HTTPException(status_code401, detail用户名或密码错误) token create_token(user.id, user.role) return {access_token: token, token_type: bearer}密码校验用pwd_context.verify它会自动从哈希串里识别当时的算法所以以后升级哈希算法不需要让老用户重新注册。登录成功后返回token_type是 OAuth2 规范的一部分Swagger 文档会自动识别这个格式方便测试。参数位置说明示例user.idpayload用户唯一标识后续查询课程、笔记都要用它1user.rolepayload区分管理员和普通用户影响课程删除权限adminexppayload过期时间戳超过后必须重新登录1700000000token_type响应体固定为bearer前端拼接 header 时用bearer依赖注入这块最实用的组合是Depends(get_db)和Depends(get_current_user)同时出现。前者负责打开数据库会话后者负责认证两者顺序无关FastAPI 会按依赖关系自动处理。当你写“获取我的课程进度”这类接口时路由函数直接拿current_user[user_id]去查表不用再读一次Authorizationheader。3. 课程模块路由、CRUD 和数据库会话边界课程模块在routers/course.py中实现表面上是增删改查实际上考验的是“数据库会话边界”。FastAPI 每个请求都应当有独立的Session请求结束就关闭否则会出现连接泄漏。这个项目的models/get_db.py就是为这件事服务的。3.1 course.py 的路由如何拆分course.py用的是APIRouter独立实例而不是把所有路由写在main.py里。好处很明显课程接口和用户接口的路径前缀、标签、依赖可以各自管理。比如课程模块需要管理员权限就单独在删除接口上加依赖而查询接口不需要。from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from models.get_db import get_db from common.jwtTool import get_current_user router APIRouter(prefix/course, tags[course]) router.get(/list) def list_courses(page: int 1, page_size: int 10, db: Session Depends(get_db)): courses get_course_page(db, page, page_size) return {total: len(courses), items: courses}prefix/course让路由在 OpenAPI 文档里统一归类tags控制 Swagger 分组Depends(get_db)传入一个可调用的生成器函数。注意get_db必须是生成器函数用yield交出 session最后再关闭。这样即使路由中途抛出异常也会走finally关闭连接。3.2 课程新增与查询的 SQLAlchemy 实现课程模型在models/model.py中定义通常长这样from sqlalchemy import Column, Integer, String, Text from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class Course(Base): __tablename__ course id Column(Integer, primary_keyTrue, indexTrue) title Column(String(255), nullableFalse) description Column(Text, default) video_url Column(String(500), default) created_by Column(Integer, indexTrue)查询操作放在models/crud.py里而不是塞进路由。这样course.py保持薄薄一层以后换数据库、改动 Join 逻辑都不影响 HTTP 层。分页查询的常见写法是offsetlimitdef get_course_page(db: Session, page: int, page_size: int): if page 1: page 1 return db.query(Course).order_by(Course.id.desc()).offset((page - 1) * page_size).limit(page_size).all()offset计算公式(page - 1) * page_size是必须记牢的page 从 1 开始page_size 决定每页数量。如果传page0SQLAlchemy 不会报错但结果会偏移到第一页之前所以要在入口处拦截。新增课程时crud.create_course需要接收created_by参数这个值就是从get_current_user依赖里拿到的用户 ID不能信任前端传值否则任何人都能冒充管理员。3.3 学习进度、笔记与用户 ID 的关联学习模块是这个项目的核心展示点进度、笔记、作业都通过user_idcourse_id联合关系落库。一个合理的进度表模型如下class CourseProgress(Base): __tablename__ course_progress id Column(Integer, primary_keyTrue) user_id Column(Integer, nullableFalse, indexTrue) course_id Column(Integer, nullableFalse, indexTrue) watched_seconds Column(Integer, default0) finished Column(Integer, default0) updated_at Column(DateTime, defaultdatetime.now, onupdatedatetime.now)更新进度的接口应该用“合并写”而不是先查再改。常见做法是先get如果不存在就新建存在就累加秒数progress db.query(CourseProgress).filter_by( user_iduser_id, course_idcourse_id ).first() if not progress: progress CourseProgress(user_iduser_id, course_idcourse_id, watched_secondsseconds) db.add(progress) else: progress.watched_seconds seconds db.commit()这里有个容易踩的坑onupdatedatetime.now是在 SQLAlchemy 执行 UPDATE 时自动触发的但如果你是先查出来再修改然后db.commit()updated_at才会遵守这个设置。如果直接执行一条底层 UPDATE 语句ORM 钩子可能不会触发。笔记和评价表的逻辑类似但查询时通常需要联表返回课程标题。这时候crud.py里会多一个join查询rows db.query(Note, Course.title).join(Course, Course.id Note.course_id).filter(Note.user_id user_id).all()返回的rows是元组列表路由层需要把它们转成字典否则 FastAPI 无法直接序列化。常见的做法是列表推导式return [{id: n.id, course_title: title, content: n.content} for n, title in rows]这一小段转换逻辑值得写在routers/course.py里而不是塞进crud.py因为crud只负责数据访问HTTP 展示层的数据整形是路由职责。4. 配置与环境config.py、双数据库文件和日志切割很多毕设项目把数据库地址、密钥、日志路径全部散落在不同文件启动时靠环境变量硬拼。这个 FastAPI 源码里单独放了一个config.py还同时保留test_database.py和prod_database.py这个设计实际上是为了解决“本地开发代码能跑一部署就连不上库”的问题。4.1 config.py 该放什么config.py更像一个集中式配置入口通常包含密钥、token 过期时间、数据库连接串和基础路径import os from pathlib import Path BASE_DIR Path(__file__).resolve().parent SECRET_KEY os.getenv(SECRET_KEY, dev-secret-change-me) TOKEN_EXPIRE_SECONDS int(os.getenv(TOKEN_EXPIRE_SECONDS, 86400)) DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./test.db)用os.getenv是为了在部署时用环境变量覆盖默认值而不是每次改代码。常见的错误是把DATABASE_URL写成硬编码的绝对路径一旦项目目录被移动或在另一台机器上运行SQLite 就会报unable to open database file。用Path(__file__).resolve().parent构造相对路径才是稳妥做法。4.2 test_database.py 与 prod_database.py 的分工models/test_database.py和models/prod_database.py名字不同内部结构其实相似都是创建 engine、声明SessionLocal、提供get_db。不同的是目标数据库文件适用场景数据库驱动常见连接串test_database.py本地开发、单元测试SQLitesqlite:///./test.dbprod_database.py服务器部署PostgreSQL / MySQLpostgresql://user:passhost:5432/db切换环境时不要同时 import 两份而是让get_db从同一个 session 工厂读取。比如get_db.py里用DATABASE_URL决定用哪个 engine这才是两个文件存在的意义。from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from config import DATABASE_URL engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close()check_same_threadFalse只在 SQLite 时需要因为 FastAPI 的线程池可能多个线程同时访问同一个连接。换成 PostgreSQL 后不需要这个参数但如果你的代码里直接复制粘贴会导致 PostgreSQL 的 psycopg2 报参数不支持错误。正确的做法是根据数据库类型动态拼接connect_args或者干脆只用DATABASE_URL判断。4.3 configLog.py 与 filepath.py 的实用写法configLog.py应该是被忽略但价值很高的文件。FastAPI 应用默认把日志打到控制台一旦后台上线你需要把请求耗时、错误堆栈写到文件里。一个可用的 RotatingFileHandler 配置如下import logging from logging.handlers import RotatingFileHandler def setup_logging(): handler RotatingFileHandler(app.log, maxBytes10_000_000, backupCount5, encodingutf-8) formatter logging.Formatter(%(asctime)s %(levelname)s %(name)s: %(message)s) handler.setFormatter(formatter) root logging.getLogger() root.addHandler(handler) root.setLevel(logging.INFO)maxBytes10_000_000即单个日志文件超过 10MB 就滚动backupCount5保留最近 5 个备份。如果项目长期运行最怕的就是日志文件无限膨胀这个配置能避免磁盘爆掉。补充一句filepath.py用来统一返回上传目录或静态资源路径比如课程封面的存储位置from pathlib import Path from config import BASE_DIR UPLOAD_DIR BASE_DIR / uploads VIDEO_DIR BASE_DIR / videos def ensure_dirs(): UPLOAD_DIR.mkdir(exist_okTrue) VIDEO_DIR.mkdir(exist_okTrue)Path.mkdir(exist_okTrue)会在目录已存在时静默跳过不会抛FileExistsError。在main.py的启动事件里调用ensure_dirs()可以保证第一次运行项目时必要的目录都存在避免上传接口因为目录不存在而 500。5. 从 Uvicorn 启动到登录接口验证源码项目拿到手后第一件事不是读代码而是先让它跑起来。我习惯把启动和验证压成固定流程五步之内能确认项目是否正常再决定往哪个方向改。5.1 安装依赖并启动服务先激活虚拟环境进入 zip 解压后的目录cd fastapi-course-system # Windows 下激活虚拟环境 venv\\Scripts\\activate # bash: source venv/bin/activate python -m pip install -r requirements.txt uvicorn main:app --host 0.0.0.0 --port 8000 --reloadmain:app表示从main.py中找到名为app的 FastAPI 实例--reload只适合开发环境修改代码后自动重启。如果你的机器上没有 venv直接安装在全局环境也能跑但容易和系统 Python 冲突。建议用python -m venv venv重建一个干净的虚拟环境不要使用 zip 里自带的那个因为原环境里的路径通常指向本机绝对路径。5.2 用 Swagger 和 TestClient 验证启动后浏览器访问http://127.0.0.1:8000/docsFastAPI 会自动生成 OpenAPI 交互文档。先点/user/register创建一个测试账号再调/user/login拿 token最后把 token 复制到页面右上角的 Authorize 对话框就可以测试课程接口。如果不想通过浏览器也可以用一个最小化的 pytest 脚本from fastapi.testclient import TestClient from main import app client TestClient(app) def test_register_and_login(): r client.post(/user/register, json{username: alice, password: 123456}) assert r.status_code 200 r client.post(/user/login, json{username: alice, password: 123456}) assert r.status_code 200 assert access_token in r.json()TestClient基于 httpx不需要真的启动 Uvicorn就能模拟完整的 HTTP 请求。上面的断言能快速发现密码哈希、数据库连接、路由注册三层问题通常作为集成测试的入口。5.3 最容易踩的三个坑第一个坑是启动时报ModuleNotFoundError: No module named jwt。这是安装的包名不匹配正确的安装命令是pip install pyjwt而不是pip install jwt两者导入名不同。第二个坑是 Pydantic v2 中orm_mode改成了model_config ConfigDict(from_attributesTrue)旧代码里如果写成class Config: orm_mode True在较新的 FastAPI 版本下会直接崩溃需要批量替换。第三个坑是 SQLite 在高并发下出现database is locked这是写入并发导致的课程学习类的读写压力不高可以先给进度表增加indexTrue字段优化如果还不够就切换到prod_database.py里的 PostgreSQL 配置。这三个问题解决后整个项目的接口链路基本不会有大毛病。最后一步建议你把test_database.py里的 SQLite 地址改成内存模式sqlite:///:memory:再用TestClient跑一遍全流程确认所有表都能在测试用临时库中重建这样才算真正把环境问题隔离干净。本文还有配套的精品资源点击获取 FastAPI在线课程系统源码解析：JWT认证与SQLAlchemy实践
简介基于FastAPI框架的在线课程学习系统是一份完整的Python项目源码与说明文档适合有一定Python基础、正在做毕设或课程设计的开发者参考也可作为在线教育平台快速原型的搭建模板。压缩包共56个文件以17个py源码文件与32个pyd编译模块为主另含依赖清单、项目说明及配置文件整体仅4.37MB目录按功能模块清晰划分便于快速定位用户、课程、学习、互动、管理等核心业务代码。系统覆盖用户注册登录、课程增删改查、学习进度跟踪、笔记作业、讨论评价和管理员后台等完整业务链能够直观看到FastAPI如何组织路由、连接数据库、封装通用工具以及处理鉴权与日志等关键环节。目前已有35人学习通过研读源码和项目说明可以掌握JWT鉴权、日志配置、分层模型等实战技巧对提升FastAPI项目开发能力有切实帮助。1. 先把 FastAPI 在线课程学习系统的源码盘顺一个毕设级别的 FastAPI 项目最容易翻车的地方不在业务代码而在目录分层和 token 处理。拆开这个 zip 压缩包你会发现main.py只负责挂载路由业务被拆到routers、common、models三个目录正是这层拆法决定了你改一个功能要动几个文件。这份基于 FastAPI 的在线课程学习 python 源码把用户注册登录、课程增删改查、学习进度和笔记都塞进了不到十个核心文件里没有分布式组件也没有消息队列核心就是 FastAPI SQLAlchemy JWT。项目适合有 Python 基础、正在做课程设计或想快速理解 FastAPI 工程化套路的开发者。下面直接从工程布局、认证链路、课程 CRUD、环境配置一路讲到 Uvicorn 启动和接口验证中间会给出能直接抄的代码和参数说明。2. 用户认证JWT 工具、注册登录与依赖注入在线课程系统里用户模块是第一个要面对的硬骨头。课程可以匿名浏览但提交笔记、记录进度、发布评价都必须知道“你是谁”。这个项目把认证抽到了common/jwtTool.py再用 FastAPI 的依赖注入把解析出来的用户信息塞进路由函数这一层设计一旦理顺后面所有业务接口都会很省事。2.1 jwtTool.pytoken 生成与解析的核心common/jwtTool.py几乎会被所有业务模块引用它负责两件事签发 token、校验 token。用 PyJWT 实现时常见做法是直接封装两个函数import time import jwt SECRET_KEY your-secret-key # 生产环境必须从 config.py 读取 ALGORITHM HS256 EXPIRE_SECONDS 60 * 60 * 24 # 默认 24 小时过期 def create_token(user_id: int, role: str user) - str: payload { user_id: user_id, role: role, exp: int(time.time()) EXPIRE_SECONDS } return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM) def parse_token(token: str) - dict: try: return jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) except jwt.ExpiredSignatureError as e: raise ValueError(token 已过期) from e except jwt.InvalidTokenError as e: raise ValueError(无效 token) from e这段代码的要点可以从三个参数看exp是标准过期声明PyJWT 会在解码时自动校验不需要手动比较时间ALGORITHM常见用 HS256同一个密钥负责签名和验证适合单体后端SECRET_KEY如果写死在代码里日后很难轮换这个项目在config.py里统一管理jwtTool.py应当只保持引用关系。实际运行时FastAPI 路由不会直接调parse_token而是把它包进一个依赖函数。比如在routers/user.py里会有类似下面的代码用Depends把解析结果注入到视图函数中from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security HTTPBearer() def get_current_user(credentials: HTTPAuthorizationCredentials Depends(security)): payload parse_token(credentials.credentials) return {user_id: payload[user_id], role: payload[role]}这里HTTPBearer会从请求头Authorization: Bearer token中自动取证书省去自己写 header 解析的工作。如果Authorization缺失FastAPI 直接返回 403路由函数里连异常判断都不需要写。2.2 注册接口密码校验与入库用户注册的核心不是“往表里插一条记录”而是先解决两个问题密码不能明文存用户名不能重复。密码处理一般用passlib或 Python 自带的hashlib。这个项目以轻量为主所以你会看到类似pbkdf2_sha256的哈希方式from passlib.context import CryptContext from fastapi import APIRouter, HTTPException from schemas import UserCreate pwd_context CryptContext(schemes[pbkdf2_sha256], deprecatedauto) router APIRouter(prefix/user, tags[user]) router.post(/register) def register(user: UserCreate): if get_user_by_username(user.username): raise HTTPException(status_code400, detail用户名已被注册) hashed_password pwd_context.hash(user.password) create_user(usernameuser.username, hashed_passwordhashed_password) return {msg: 注册成功}UserCreate是 Pydantic 模型至少要校验username和password字段非空。注意这里必须返回明确的中文错误提示否则前端很难把“409/400”翻译成对用户友好的文案。get_user_by_username和create_user在models/crud.py中实现实际上就是 SQLAlchemy 的session.query与add/commit。这里有个经常被忽略的细节注册接口不要返回敏感字段。干脆用 Pydantic 的响应模型只暴露id、username、created_at。否则一旦后来加了hashed_password字段会直接泄露哈希。models/schemas.py里一般会区分UserCreate、UserOut两种模型这就是 FastAPI 推荐的项目结构。2.3 登录与依赖注入获取当前用户登录接口和注册不同它不做“创建”而是根据用户名查出用户、校验密码、签发 token。一个能直接用的写法如下router.post(/login) def login(form: UserLogin, session: Session Depends(get_db)): user get_user_by_username(session, form.username) if not user or not pwd_context.verify(form.password, user.hashed_password): raise HTTPException(status_code401, detail用户名或密码错误) token create_token(user.id, user.role) return {access_token: token, token_type: bearer}密码校验用pwd_context.verify它会自动从哈希串里识别当时的算法所以以后升级哈希算法不需要让老用户重新注册。登录成功后返回token_type是 OAuth2 规范的一部分Swagger 文档会自动识别这个格式方便测试。参数位置说明示例user.idpayload用户唯一标识后续查询课程、笔记都要用它1user.rolepayload区分管理员和普通用户影响课程删除权限adminexppayload过期时间戳超过后必须重新登录1700000000token_type响应体固定为bearer前端拼接 header 时用bearer依赖注入这块最实用的组合是Depends(get_db)和Depends(get_current_user)同时出现。前者负责打开数据库会话后者负责认证两者顺序无关FastAPI 会按依赖关系自动处理。当你写“获取我的课程进度”这类接口时路由函数直接拿current_user[user_id]去查表不用再读一次Authorizationheader。3. 课程模块路由、CRUD 和数据库会话边界课程模块在routers/course.py中实现表面上是增删改查实际上考验的是“数据库会话边界”。FastAPI 每个请求都应当有独立的Session请求结束就关闭否则会出现连接泄漏。这个项目的models/get_db.py就是为这件事服务的。3.1 course.py 的路由如何拆分course.py用的是APIRouter独立实例而不是把所有路由写在main.py里。好处很明显课程接口和用户接口的路径前缀、标签、依赖可以各自管理。比如课程模块需要管理员权限就单独在删除接口上加依赖而查询接口不需要。from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from models.get_db import get_db from common.jwtTool import get_current_user router APIRouter(prefix/course, tags[course]) router.get(/list) def list_courses(page: int 1, page_size: int 10, db: Session Depends(get_db)): courses get_course_page(db, page, page_size) return {total: len(courses), items: courses}prefix/course让路由在 OpenAPI 文档里统一归类tags控制 Swagger 分组Depends(get_db)传入一个可调用的生成器函数。注意get_db必须是生成器函数用yield交出 session最后再关闭。这样即使路由中途抛出异常也会走finally关闭连接。3.2 课程新增与查询的 SQLAlchemy 实现课程模型在models/model.py中定义通常长这样from sqlalchemy import Column, Integer, String, Text from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class Course(Base): __tablename__ course id Column(Integer, primary_keyTrue, indexTrue) title Column(String(255), nullableFalse) description Column(Text, default) video_url Column(String(500), default) created_by Column(Integer, indexTrue)查询操作放在models/crud.py里而不是塞进路由。这样course.py保持薄薄一层以后换数据库、改动 Join 逻辑都不影响 HTTP 层。分页查询的常见写法是offsetlimitdef get_course_page(db: Session, page: int, page_size: int): if page 1: page 1 return db.query(Course).order_by(Course.id.desc()).offset((page - 1) * page_size).limit(page_size).all()offset计算公式(page - 1) * page_size是必须记牢的page 从 1 开始page_size 决定每页数量。如果传page0SQLAlchemy 不会报错但结果会偏移到第一页之前所以要在入口处拦截。新增课程时crud.create_course需要接收created_by参数这个值就是从get_current_user依赖里拿到的用户 ID不能信任前端传值否则任何人都能冒充管理员。3.3 学习进度、笔记与用户 ID 的关联学习模块是这个项目的核心展示点进度、笔记、作业都通过user_idcourse_id联合关系落库。一个合理的进度表模型如下class CourseProgress(Base): __tablename__ course_progress id Column(Integer, primary_keyTrue) user_id Column(Integer, nullableFalse, indexTrue) course_id Column(Integer, nullableFalse, indexTrue) watched_seconds Column(Integer, default0) finished Column(Integer, default0) updated_at Column(DateTime, defaultdatetime.now, onupdatedatetime.now)更新进度的接口应该用“合并写”而不是先查再改。常见做法是先get如果不存在就新建存在就累加秒数progress db.query(CourseProgress).filter_by( user_iduser_id, course_idcourse_id ).first() if not progress: progress CourseProgress(user_iduser_id, course_idcourse_id, watched_secondsseconds) db.add(progress) else: progress.watched_seconds seconds db.commit()这里有个容易踩的坑onupdatedatetime.now是在 SQLAlchemy 执行 UPDATE 时自动触发的但如果你是先查出来再修改然后db.commit()updated_at才会遵守这个设置。如果直接执行一条底层 UPDATE 语句ORM 钩子可能不会触发。笔记和评价表的逻辑类似但查询时通常需要联表返回课程标题。这时候crud.py里会多一个join查询rows db.query(Note, Course.title).join(Course, Course.id Note.course_id).filter(Note.user_id user_id).all()返回的rows是元组列表路由层需要把它们转成字典否则 FastAPI 无法直接序列化。常见的做法是列表推导式return [{id: n.id, course_title: title, content: n.content} for n, title in rows]这一小段转换逻辑值得写在routers/course.py里而不是塞进crud.py因为crud只负责数据访问HTTP 展示层的数据整形是路由职责。4. 配置与环境config.py、双数据库文件和日志切割很多毕设项目把数据库地址、密钥、日志路径全部散落在不同文件启动时靠环境变量硬拼。这个 FastAPI 源码里单独放了一个config.py还同时保留test_database.py和prod_database.py这个设计实际上是为了解决“本地开发代码能跑一部署就连不上库”的问题。4.1 config.py 该放什么config.py更像一个集中式配置入口通常包含密钥、token 过期时间、数据库连接串和基础路径import os from pathlib import Path BASE_DIR Path(__file__).resolve().parent SECRET_KEY os.getenv(SECRET_KEY, dev-secret-change-me) TOKEN_EXPIRE_SECONDS int(os.getenv(TOKEN_EXPIRE_SECONDS, 86400)) DATABASE_URL os.getenv(DATABASE_URL, sqlite:///./test.db)用os.getenv是为了在部署时用环境变量覆盖默认值而不是每次改代码。常见的错误是把DATABASE_URL写成硬编码的绝对路径一旦项目目录被移动或在另一台机器上运行SQLite 就会报unable to open database file。用Path(__file__).resolve().parent构造相对路径才是稳妥做法。4.2 test_database.py 与 prod_database.py 的分工models/test_database.py和models/prod_database.py名字不同内部结构其实相似都是创建 engine、声明SessionLocal、提供get_db。不同的是目标数据库文件适用场景数据库驱动常见连接串test_database.py本地开发、单元测试SQLitesqlite:///./test.dbprod_database.py服务器部署PostgreSQL / MySQLpostgresql://user:passhost:5432/db切换环境时不要同时 import 两份而是让get_db从同一个 session 工厂读取。比如get_db.py里用DATABASE_URL决定用哪个 engine这才是两个文件存在的意义。from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from config import DATABASE_URL engine create_engine(DATABASE_URL, connect_args{check_same_thread: False}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close()check_same_threadFalse只在 SQLite 时需要因为 FastAPI 的线程池可能多个线程同时访问同一个连接。换成 PostgreSQL 后不需要这个参数但如果你的代码里直接复制粘贴会导致 PostgreSQL 的 psycopg2 报参数不支持错误。正确的做法是根据数据库类型动态拼接connect_args或者干脆只用DATABASE_URL判断。4.3 configLog.py 与 filepath.py 的实用写法configLog.py应该是被忽略但价值很高的文件。FastAPI 应用默认把日志打到控制台一旦后台上线你需要把请求耗时、错误堆栈写到文件里。一个可用的 RotatingFileHandler 配置如下import logging from logging.handlers import RotatingFileHandler def setup_logging(): handler RotatingFileHandler(app.log, maxBytes10_000_000, backupCount5, encodingutf-8) formatter logging.Formatter(%(asctime)s %(levelname)s %(name)s: %(message)s) handler.setFormatter(formatter) root logging.getLogger() root.addHandler(handler) root.setLevel(logging.INFO)maxBytes10_000_000即单个日志文件超过 10MB 就滚动backupCount5保留最近 5 个备份。如果项目长期运行最怕的就是日志文件无限膨胀这个配置能避免磁盘爆掉。补充一句filepath.py用来统一返回上传目录或静态资源路径比如课程封面的存储位置from pathlib import Path from config import BASE_DIR UPLOAD_DIR BASE_DIR / uploads VIDEO_DIR BASE_DIR / videos def ensure_dirs(): UPLOAD_DIR.mkdir(exist_okTrue) VIDEO_DIR.mkdir(exist_okTrue)Path.mkdir(exist_okTrue)会在目录已存在时静默跳过不会抛FileExistsError。在main.py的启动事件里调用ensure_dirs()可以保证第一次运行项目时必要的目录都存在避免上传接口因为目录不存在而 500。5. 从 Uvicorn 启动到登录接口验证源码项目拿到手后第一件事不是读代码而是先让它跑起来。我习惯把启动和验证压成固定流程五步之内能确认项目是否正常再决定往哪个方向改。5.1 安装依赖并启动服务先激活虚拟环境进入 zip 解压后的目录cd fastapi-course-system # Windows 下激活虚拟环境 venv\\Scripts\\activate # bash: source venv/bin/activate python -m pip install -r requirements.txt uvicorn main:app --host 0.0.0.0 --port 8000 --reloadmain:app表示从main.py中找到名为app的 FastAPI 实例--reload只适合开发环境修改代码后自动重启。如果你的机器上没有 venv直接安装在全局环境也能跑但容易和系统 Python 冲突。建议用python -m venv venv重建一个干净的虚拟环境不要使用 zip 里自带的那个因为原环境里的路径通常指向本机绝对路径。5.2 用 Swagger 和 TestClient 验证启动后浏览器访问http://127.0.0.1:8000/docsFastAPI 会自动生成 OpenAPI 交互文档。先点/user/register创建一个测试账号再调/user/login拿 token最后把 token 复制到页面右上角的 Authorize 对话框就可以测试课程接口。如果不想通过浏览器也可以用一个最小化的 pytest 脚本from fastapi.testclient import TestClient from main import app client TestClient(app) def test_register_and_login(): r client.post(/user/register, json{username: alice, password: 123456}) assert r.status_code 200 r client.post(/user/login, json{username: alice, password: 123456}) assert r.status_code 200 assert access_token in r.json()TestClient基于 httpx不需要真的启动 Uvicorn就能模拟完整的 HTTP 请求。上面的断言能快速发现密码哈希、数据库连接、路由注册三层问题通常作为集成测试的入口。5.3 最容易踩的三个坑第一个坑是启动时报ModuleNotFoundError: No module named jwt。这是安装的包名不匹配正确的安装命令是pip install pyjwt而不是pip install jwt两者导入名不同。第二个坑是 Pydantic v2 中orm_mode改成了model_config ConfigDict(from_attributesTrue)旧代码里如果写成class Config: orm_mode True在较新的 FastAPI 版本下会直接崩溃需要批量替换。第三个坑是 SQLite 在高并发下出现database is locked这是写入并发导致的课程学习类的读写压力不高可以先给进度表增加indexTrue字段优化如果还不够就切换到prod_database.py里的 PostgreSQL 配置。这三个问题解决后整个项目的接口链路基本不会有大毛病。最后一步建议你把test_database.py里的 SQLite 地址改成内存模式sqlite:///:memory:再用TestClient跑一遍全流程确认所有表都能在测试用临时库中重建这样才算真正把环境问题隔离干净。本文还有配套的精品资源点击获取