news 2026/9/15 20:01:32

FastAPI实战:构建带JWT登录与断点续学的在线课程学习系统

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
FastAPI实战:构建带JWT登录与断点续学的在线课程学习系统

简介:基于FastAPI框架的在线课程学习系统Python源码及项目说明,是一份面向具备一定Python基础的开发者的完整参考实现。系统以在线学习平台为场景,围绕用户、课程、学习、互动、管理五大模块构建,覆盖用户注册登录与信息维护、课程添加删除修改查询、学习进度跟踪、笔记记录、作业提交、讨论答疑、评价以及管理员后台管理等功能,功能链条完整,适合用于毕业设计、课程设计或项目练手。压缩包共56个文件,包含大量.py源码、.pyd编译模块、.pyc缓存文件及README说明文档,并附有requirements.txt运行依赖与必要配置文件,整体体积仅4.37MB,目录划分清晰,便于按模块查阅。目前已有35人学习下载。通过阅读源码与项目说明,可以掌握FastAPI路由划分、JWT认证、数据库CRUD交互、日志配置等关键实现,也能借鉴多模块系统拆分思路与工程组织方式,对提升后端开发能力和完成同类系统设计均有实际帮助。

1. 基于FastAPI框架的在线课程学习系统到底在解决什么问题

一个基于FastAPI框架的在线课程学习系统,表面看起来只是课程表、用户、视频链接的管理后台,但真正难的是“学习进度”状态:看到第几章、停在哪一秒、测验过没过,跨设备同步比课程表本身复杂。FastAPI 适合这个场景,类型注解同时做参数校验和接口文档,异步能力让课件上传、学习记录写入不阻塞主流程。

这里讲的“基于 FastAPI 框架的在线课程学习系统 python 源码 + 项目说明”,通常包含课程管理、用户注册登录、学习进度记录、静态课件托管,再加一份能跑起来的说明。拿到这种 python 源码时,先看数据模型、鉴权方式和依赖清单,比急着重装环境更有用。

适合想用 Python 写业务后端、交付毕设或内部平台的人,也适合跟着 FastAPI 教程到一半、想找一个完整项目实战的读者。下面按环境模型、核心接口、页面资源、打包验证四层展开。

2. 先立技术底座:FastAPI项目骨架与数据模型

2.1 用uv还是pip创建FastAPI开发环境?最小命令先跑通

Python 后端项目交付时最常见的翻车点不是代码逻辑,而是依赖环境。同一个 FastAPI 项目,换一台电脑后因为 uvicorn、pydantic 或 SQLAlchemy 版本漂移而跑不起来,几乎每个人都遇到过。所以在线课程学习系统这类“源码+说明”项目,第一步就是把依赖环境固定住。我现在默认用 uv,它创建虚拟环境和解析依赖都比 pip 快,还会生成uv.lock,让“解压就能跑”更现实。

在空目录执行:

uv init fastapi-course cd fastapi-course uv add fastapi "uvicorn[standard]" sqlalchemy pydantic-settings python-jose[cryptography] passlib[bcrypt] python-multipart

没有 uv 时,等价的 pip 命令是:

python -m venv .venv source .venv/bin/activate # Windows PowerShell 用 .venv\Scripts\Activate.ps1 pip install fastapi "uvicorn[standard]" sqlalchemy pydantic-settings python-jose[cryptography] passlib[bcrypt] python-multipart

参数说明:uvicorn[standard]会带上 uvloop、websockets、httptools,让开发服务器在高并发下表现更好;pydantic-settings负责从.env读配置;python-jose[cryptography]用来签发和验证 JWT;passlib[bcrypt]做密码散列;python-multipart是 FastAPI 解析表单和文件上传的必要依赖,不装的话UploadFile接口调用时直接报Form data requires "python-multipart"

“pycharm 安装 fastapi 失败报错”是搜索量很高的词。这个问题多数不是 FastAPI 装不上,而是 PyCharm 选了 conda 基础环境,权限不够或镜像源不通。更稳的做法是先建好.venv,然后在 PyCharm 里选择这个解释器,再在 Terminal 中执行uv sync。命令行和 IDE 环境一致,比在包管理面板里反复点安装更可控。

2.2 数据模型怎么拆:课程、章节、用户、学习进度四张表

在线课程学习系统的表结构不能拍脑袋合并。常见做法是四张业务表加一张选课关联表:courses放课程元信息,sections放章节,users放账号,course_progress放学习进度,user_course记录选课关系。

职责最常见的问题
courses课程标题、简介、封面、是否上架把视频地址直接写进课程字段,后续无法扩展多版本
sections章节标题、视频地址、排序删除课程时没同步删章节,出现孤儿数据
users用户名、密码散列、角色密码明文存储,安全评审直接被退回
course_progress用户、课程、播放位置、时长缺联合唯一约束,产生重复进度
user_course选课关系没建(user_id, course_id)唯一索引

SQLAlchemy 2.x 声明式模型可以这样写:

from sqlalchemy import ForeignKey, String, Text, UniqueConstraint from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class Course(Base): __tablename__ = "courses" id: Mapped[int] = mapped_column(primary_key=True) title: Mapped[str] = mapped_column(String(200), index=True) description: Mapped[str] = mapped_column(Text, default="") is_published: Mapped[bool] = mapped_column(default=False) sections: Mapped[list["Section"]] = relationship( back_populates="course", cascade="all, delete-orphan" ) class Section(Base): __tablename__ = "sections" __table_args__ = (UniqueConstraint("course_id", "sort_order"),) id: Mapped[int] = mapped_column(primary_key=True) course_id: Mapped[int] = mapped_column(ForeignKey("courses.id", ondelete="CASCADE")) title: Mapped[str] = mapped_column(String(200)) video_url: Mapped[str] = mapped_column(String(500)) sort_order: Mapped[int] = mapped_column(default=0) course: Mapped[Course] = relationship(back_populates="sections")

参数说明:UniqueConstraint("course_id", "sort_order")防止同一课程下出现两个“第 1 讲”,比在业务代码里先查再插更可靠。cascade="all, delete-orphan"让删除课程时自动清理章节,避免留下外键引用。注意 SQLite 默认不会真正执行外键约束,初始化连接后最好执行PRAGMA foreign_keys=ON,这个细节最容易在本地环境漏掉。

2.3 项目启动时读配置:pydantic_settings、上传目录和数据库初始化

FastAPI 初始化配置不要散落在路由文件里。官方周边里负责这件事的是 pydantic-settings,它能把配置集中在一个Settings类里,并支持环境变量覆盖.env文件。在线课程系统要部署到不同环境时,只改DATABASE_URL就够了。

from pathlib import Path from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): database_url: str = "sqlite:///./course.db" secret_key: str = "please-change-me" access_token_expire_minutes: int = 60 * 24 upload_dir: str = "uploads" model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", extra="ignore", ) settings = Settings()

应用入口用 lifespan 初始化目录:

from contextlib import asynccontextmanager from pathlib import Path from fastapi import FastAPI from app.config import settings @asynccontextmanager async def lifespan(app: FastAPI): Path(settings.upload_dir).mkdir(parents=True, exist_ok=True) # 数据库建表或迁移放到这里,不要写在路由函数里 yield app = FastAPI(title="在线课程学习系统", lifespan=lifespan)

参数说明:extra="ignore".env里出现模板没有的变量时不启动报错,开发和生产的配置可以不一样。database_url默认给 SQLite,交付后不需要先装 MySQL;团队要用 MySQL 时,只需在.env覆盖DATABASE_URL=mysql+pymysql://user:pass@host/course。把配置、启动、表结构固定好,后面的接口才有一个稳定的底座。

这里有一点要提醒:SQLAlchemy 模型和 Pydantic 模型是两套东西,不要混用一个类。FastAPI 利用类型注解生成文档和校验,接口层再用 Pydantic 定义请求体和返回模型,会清晰很多。

3. 在线课程系统核心API实现:从JWT登录到断点续学

3.1 用户注册与登录:FastAPI里OAuth2PasswordBearer怎么接JWT

在线课程系统再小,也要有登录。它不只是为限制访问,而是学习进度必须绑定一个用户。最稳妥的方案是 JWT Bearer 认证:用户拿密码换 token,之后每个请求带Authorization: Bearer <token>

密码处理先单独看:

from datetime import datetime, timedelta, timezone import jwt from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") SECRET_KEY = "please-change-me" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def create_access_token(subject: str | int) -> str: expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) payload = {"sub": str(subject), "exp": expire} return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

参数说明:sub在 JWT 规范里要求是字符串,所以用str(subject)包一层;exp用 UTC 时间,避免服务器时区不一致导致 token 提前失效;CryptContext会自动处理每次散列的盐值,不要自己拼盐。

登录路由:

from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login") @app.post("/api/auth/login") def login(form: OAuth2PasswordRequestForm = Depends()): user = user_service.get_by_username(form.username) if not user or not verify_password(form.password, user.hashed_password): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误") return {"access_token": create_access_token(user.id), "token_type": "bearer"}

OAuth2PasswordRequestForm是 FastAPI 内置的表单依赖,自动从请求体里取usernamepassword,Swagger UI 的 Authorize 按钮可以直接用。tokenUrl只用来告诉 OpenAPI 文档“token 在该地址获取”,不负责跳转。后续路由用Depends(oauth2_scheme)拿 token,再解析当前用户。这里有个坑:只解 token 不查库,用户被禁用或删除后,token 在过期前依然能访问接口,所以在线课程系统至少要做一次数据库回查。

JWT 相关报错比较常见,整理成排查表:

报错或现象常见原因排查方向
module 'bcrypt' has no attribute '__about__'passlib 与 bcrypt 版本不兼容固定bcrypt==4.0.1
Swagger UI 点 Authorize 没反应tokenUrl路径写错openapi.json里的 oauth2 配置
登录接口 401密码校验失败或用户不存在先手工调用verify_password验散列
token 解析报exp错误服务器时间偏差大同步 NTP 或统一使用 UTC

3.2 课程与章节接口:分页、关键字、发布状态怎么控制

课程列表是所有在线课程系统前面的接口,它要同时照顾学生和教师:学生只能看已发布课程,教师能看全部。接口上一般用Depends先取当前用户,再根据角色决定过滤条件。

from fastapi import Depends, Query from sqlalchemy.orm import Session @app.get("/api/courses") def list_courses( keyword: str | None = Query(default=None, max_length=50), page: int = Query(default=1, ge=1), page_size: int = Query(default=10, ge=1, le=50), db: Session = Depends(get_db), user: User = Depends(get_current_user), ): query = db.query(Course).filter(Course.is_published.is_(True)) if keyword: query = query.filter(Course.title.contains(keyword)) total = query.count() items = query.order_by(Course.id.desc()).offset((page - 1) * page_size).limit(page_size).all() return {"total": total, "items": items, "page": page, "page_size": page_size}

参数里Query(ge=1, le=50)是 FastAPI 在路由层做的边界校验,page=0page_size=10000这类请求会在进入业务逻辑前被拦截并返回 422。max_length=50控制关键词长度,避免超长字符串影响数据库查询。如果教师端用 FastAPI + Layui 做管理表单,分页参数是pagelimit,后端返回items后前端用 Layui 的parseData映射即可;如果是 FastAPI + Vue3 的前后端分离页面,返回标准 JSON 也同样适配,不用为不同前端改接口。

课程详情接口建议一次返回章节列表,而不是让前端循环请求每个章节:

@app.get("/api/courses/{course_id}") def get_course_detail(course_id: int, db: Session = Depends(get_db)): course = db.get(Course, course_id) if not course or not course.is_published: raise HTTPException(status_code=404, detail="课程不存在或未上架") return { "id": course.id, "title": course.title, "sections": [ {"id": s.id, "title": s.title, "video_url": s.video_url, "sort_order": s.sort_order} for s in course.sections ], }

db.get(Course, course_id)按主键查,能少一次 where 查询。访问course.sections时 SQLAlchemy 默认惰性加载,在线课程系统章节一般不多,可以接受;如果单课程章节超过几十个,建议在查询时加selectinload(Course.sections)预加载,避免 N+1 查询。

3.3 学习进度提交与断点续学:为什么不直接覆盖当前播放位置

FastAPI 接口的 Python 后端里,学习进度是业务陷阱最多的接口。前端页面会周期性上报“当前播放到第 150 秒”,如果后端每次直接row.position = payload.position,那用户回退观看时,已学到的较远进度会被后面的回调覆盖;乱序请求到达时,最终位置也可能不对。

我常用的做法是秒级上报时先取历史最大值,同时在关联表加duration_seconds字段,用来判断“看完”。判断完成不能只靠position == duration,因为视频可能被拖到结尾却没真正播放,一般再加一个 90% 阈值。

进度更新路由:

from datetime import datetime, timezone from pydantic import BaseModel, Field class ProgressIn(BaseModel): position_seconds: int = Field(ge=0) duration_seconds: int = Field(ge=0) @app.post("/api/courses/{course_id}/progress") def update_progress( course_id: int, payload: ProgressIn, db: Session = Depends(get_db), user: User = Depends(get_current_user), ): row = ( db.query(CourseProgress) .filter(CourseProgress.user_id == user.id, CourseProgress.course_id == course_id) .first() ) if row is None: row = CourseProgress(user_id=user.id, course_id=course_id, position_seconds=0, duration_seconds=0) db.add(row) row.position_seconds = max(row.position_seconds, payload.position_seconds) row.duration_seconds = max(row.duration_seconds, payload.duration_seconds) row.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(row) return {"position_seconds": row.position_seconds, "duration_seconds": row.duration_seconds}

参数说明:Field(ge=0)让负数在路由层就被拦截;max保证回退观看不会覆盖后面的进度;db.refresh(row)把数据库默认值刷进内存,避免返回的updated_at是 None。断点续学查询时再单独提供一个 GET 接口,前端打开课程时把视频初始化到上次位置。

在线课程系统的核心价值在这里:不是接口写得多,而是“用户下次打开还是上次的位置”。建议把进度写入放到独立接口,而不是塞进课程详情接口,否则播放器每隔几秒传一次课程信息,IO 压力会无谓增大。

4. 页面与资源:FastAPI托管静态文件、上传课件与后台任务

4.1 模板渲染还是前后端分离?根据项目说明里的交付物决定

FastAPI 前端接入有两条路:一种是用 Jinja2 模板把页面渲染在服务端,另一种是只提供 JSON API,前端用 Vue3、React 或 Layui 单独部署。在线课程学习系统选哪种,不取决于哪个更高级,而取决于交付物约束。

如果源码包里有templates目录,就用 Jinja2Templates:

from fastapi.templating import Jinja2Templates from fastapi import Request templates = Jinja2Templates(directory="templates") @app.get("/") def home(request: Request, db: Session = Depends(get_db), user: User = Depends(get_current_user)): courses = db.query(Course).filter(Course.is_published.is_(True)).all() return templates.TemplateResponse(request, "index.html", {"courses": courses})

新版TemplateResponse第一个参数是request,旧写法在 Starlette 新版会告警,网上老教程容易误导。如果项目说明写的是前端工程独立,后端只需要保证 API 地址可配。FastAPI 开发时加 CORS 即可:

from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173"], # Vue3 开发服务器默认端口 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )

生产环境不要把allow_origins写成*,否则任意网站都能读取课程和用户进度。Vue3 构建后通常有vite.config.js的路径转发配置,这时后端甚至不需要 CORS,因为前端请求走同源后端。FastAPI + Vue3 联调最容易出的问题不是 CORS,而是路径前缀不一致:前端传/api/courses,后端路由却在/courses,所以项目说明里明确 baseURL 很重要。

资源托管和异步任务选型可以先用一张表定方向:

场景推荐方案原因
服务端渲染页面Jinja2Templates交付物包含 templates 目录
独立前端工程JSON API + CORSVue3 / Layui 各自构建
用户上传课件UploadFile + uuid 改名避免路径穿越和文件名冲突
几秒内的通知任务BackgroundTasks不引入消息队列,部署简单
需重试和调度的任务Celery / RQBackgroundTasks 不保证重试

4.2 用StaticFiles暴露课程课件:上传目录和静态目录要分开

在线课程系统不可避免地要放视频封面、课件 PDF、甚至录播视频。FastAPI 里暴露目录的标准做法是app.mount

from fastapi.staticfiles import StaticFiles app.mount("/static", StaticFiles(directory="static"), name="static") app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")

/static放前端自己的 JS/CSS,/uploads放用户上传的课件,两者分开。不要把所有文件都塞进static,否则静态目录一旦允许列举,内部文件就暴露了。

上传接口要考虑角色、扩展名和文件大小:

import uuid from pathlib import Path from fastapi import File, HTTPException, UploadFile ALLOWED_EXTENSIONS = {"pdf", "mp4", "png", "jpg", "jpeg"} MAX_FILE_SIZE = 200 * 1024 * 1024 # 200MB @app.post("/api/upload") async def upload_course_file( file: UploadFile = File(...), user: User = Depends(get_current_user), ): if user.role != "teacher": raise HTTPException(status_code=403, detail="仅教师可上传课件") file_ext = file.filename.rsplit(".", 1)[-1].lower() if file_ext not in ALLOWED_EXTENSIONS: raise HTTPException(status_code=400, detail="不支持的文件类型") data = await file.read() if len(data) > MAX_FILE_SIZE: raise HTTPException(status_code=413, detail="文件不能超过200MB") unique_name = f"{uuid.uuid4().hex}.{file_ext}" Path(settings.upload_dir).joinpath(unique_name).write_bytes(data) return {"url": f"/uploads/{unique_name}", "filename": file.filename}

参数说明:File(...)表示该字段必填;await file.read()把整个文件读进内存,200MB 以下没问题,更大的文件要改成流式写盘。uuid.uuid4().hex防止文件名冲突,也避免原始文件名里的特殊字符影响保存路径。上传后返回的 URL 可直接放在<video>src上,因为访问/uploads/xxx.mp4由 StaticFiles 直接返回,不再经过业务逻辑。如果课件是付费内容,就不要用裸目录,要改用FileResponse配合登录校验返回文件。

4.3 后台任务:用BackgroundTasks生成学习报告,先别急着上Celery

课程完成时生成学习报告、清理临时文件这类操作,不需要一个完整的 Celery 集群。FastAPI 内置的BackgroundTasks适合执行时间几秒以内的任务,它会在响应返回后执行,不拖慢接口响应。

import time from fastapi import BackgroundTasks def generate_progress_report(username: str, course_title: str): with open("reports.log", "a", encoding="utf-8") as f: f.write(f"{username} 完成 {course_title}\n") time.sleep(1) @app.post("/api/courses/{course_id}/complete", status_code=202) async def complete_course(course_id: int, background_tasks: BackgroundTasks, user: User = Depends(get_current_user)): course = db.get(Course, course_id) if not course: raise HTTPException(status_code=404, detail="课程不存在") background_tasks.add_task(generate_progress_report, user.username, course.title) return {"status": "accepted", "message": "学习报告后台生成中"}

状态码202 Accepted200 OK更贴合语义,表示请求已受理但结果还没出来。add_task按顺序传函数参数即可。BackgroundTasks 能覆盖发邮件、写日志、清理临时文件;如果任务需要消息队列、失败重试或定时调度,再考虑 Celery 或 RQ。很多 FastAPI 项目实战一上来就引 Celery,反而让项目说明变长、部署变复杂,学习报告这样的轻任务完全没必要。

5. 让“源码+项目说明”可复现:压缩包里的工程纪律

5.1 压缩包目录约定

给别人的源码包必须有可预判的目录结构。下面这套约定也是我拿到“fastapi 课程系统源码”后先核对清单,逐项确认:

路径作用
backend/app/main.pyFastAPI 应用入口
backend/app/models.pySQLAlchemy 模型
backend/app/schemas.pyPydantic 请求/响应模型
backend/app/api/按业务拆分的路由
backend/requirements.txt依赖清单
backend/.env.example配置样例
backend/README.md项目说明

依赖锁定不要用pip freeze > requirements.txt一把梭,它会把当前环境里不属于项目的包也写进去,下次安装要么多余要么冲突。用 uv 导出更干净,只保留项目声明的依赖:

uv export --no-hashes -o requirements.txt

5.2 项目说明里必须写清楚哪四件事

README 最缺的通常是 Python 版本、启动命令、数据库初始化和默认账号。在线课程系统至少写下这段:

python -m venv .venv source .venv/bin/activate pip install -r requirements.txt python -m app.init_db uvicorn app.main:app --reload --port 8000

python -m app.init_db应由独立脚本实现,不要写在main.py的 import 阶段。README 还要写明测试账号和教师/学生角色的区别,否则拿到包的人只能注册学生,看不到教师管理接口。

5.3 发版前的验证命令

压缩包发出去之前,按这几条命令跑一遍,能拦住大多数依赖类问题:

python -c "import fastapi, sqlalchemy, jose, passlib; print('deps ok')" uvicorn app.main:app --reload --port 8000 curl http://localhost:8000/docs -I

curl检查/docs返回 200,说明 OpenAPI 文档正常生成。然后打开 Swagger UI 依次测注册、登录、上传 PDF、提交学习进度、重新登录看进度是否还在。这五步全跑通,“基于 FastAPI 框架的在线课程学习系统”的源码包才算真正可交付。

本文还有配套的精品资源,点击获取

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/15 20:00:03

MATLAB上采样与下采样实战:BPSK链路采样率转换深度解析

简介&#xff1a;资源聚焦数字通信中的BPSK调制与采样率变换&#xff0c;面向信号处理、通信原理方向的学生及MATLAB开发者&#xff0c;提供可直接运行的上采样与下采样实现。BPSK通过载波相位传递0/1信息&#xff0c;上采样在序列中插零提高采样率&#xff0c;下采样则按整数因…

作者头像 李华
网站建设 2026/9/15 19:59:14

Maven构建复杂项目

最近研究开源监控平台Hertzbeat&#xff0c;于是github上fork了这个项目&#xff08;master版本日期20251120&#xff09;&#xff0c;把代码拉到本地跑起来。Hertzbeat属于父子模块项目&#xff0c;通过前后端进行部署。Maven构建时软件版本要求Maven3&#xff0c;JDK17及以上…

作者头像 李华