在开始构建档案教育服务平台之前,我们需要确保本地开发环境已经配置妥当。本指南将使用 Python 作为开发语言,基于 FastAPI 框架构建高性能的 RESTful API,并使用 SQLite 作为轻量级数据库,确保零门槛快速落地。
请在终端中检查 Python 版本,确保安装了 Python 3.8 或更高版本:
python --version
接下来,我们在本地创建一个项目目录,并进入该目录:
mkdir archive_edu_service
cd archive_edu_service
为了隔离项目依赖,我们需要创建一个虚拟环境。在 Windows 系统下执行:
python -m venv venv
在 macOS 或 Linux 系统下执行:
python3 -m venv venv
创建完成后,激活虚拟环境。Windows 激活命令:
venv\Scripts\activate
macOS 或 Linux 激活命令:
source venv/bin/activate
激活成功后,命令行提示符前通常会显示 (venv) 标识。现在,我们需要安装核心依赖包。我们将使用 fastapi 作为 Web 框架,uvicorn 作为 ASGI 服务器,sqlalchemy 用于数据库 ORM 操作,以及 pydantic 用于数据验证(FastAPI 内置依赖)。请执行以下安装命令:
pip install fastapi uvicorn sqlalchemy
为了方便后续管理,建议将依赖列表导出到文件中:
pip freeze > requirements.txt
我们将构建一个基础的“档案教育”数据模型,用于存储教育档案的元数据。在本示例中,我们设计一个 EducationArchive 模型,包含档案标题、讲师、分类、描述以及创建时间等字段。
在项目根目录下创建一个名为 database.py 的文件,用于配置数据库连接。我们将使用 SQLite,因为它无需安装额外的数据库服务,文件即数据库,非常适合快速原型开发。写入以下完整代码:
```python
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL 是数据库连接地址
这里使用 sqlite://,表示创建一个名为 ./arch_edu.db 的本地文件数据库
SQLALCHEMY_DATABASE_URL = "sqlite:///./arch_edu.db"
create_engine 负责建立与数据库的连接
connect_args={"check_same_thread": False} 是 SQLite 特有的配置,允许多线程访问
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal 用于创建数据库会话,每次请求操作数据库时都需要一个独立的会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base 是所有 ORM 模型的基类
Base = declarative_base()
```
接着,创建一个名为 models.py 的文件,定义具体的数据库表结构。我们需要引入刚才配置的 Base,并定义字段类型:
```python
from sqlalchemy import Column, Integer, String, DateTime, Text
from datetime import datetime
from database import Base
class EducationArchive(Base):
__tablename__ = "education_archives"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(200), nullable=False, comment="档案标题")
instructor = Column(String(100), nullable=False, comment="讲师姓名")
category = Column(String(50), index=True, comment="档案分类")
description = Column(Text, nullable=True, comment="详细描述")
file_url = Column(String(500), nullable=True, comment="资源文件链接")
created_at = Column(DateTime, default=datetime.utcnow, comment="创建时间")
```
FastAPI 使用 Pydantic 来进行数据的序列化和反序列化验证。我们需要定义用于接收前端传入数据的 Schema(请求体)以及用于返回给前端数据的 Schema(响应体)。
创建一个名为 schemas.py 的文件,写入以下代码:
```python
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
用于创建档案时的请求模型,不需要包含 id 和 created_at
class ArchiveCreate(BaseModel):
title: str = Field(..., max_length=200, description="档案标题")
instructor: str = Field(..., max_length=100, description="讲师姓名")
category: str = Field(..., max_length=50, description="档案分类")
description: Optional[str] = Field(None, description="详细描述")
file_url: Optional[str] = Field(None, max_length=500, description="资源链接")
用于返回给前端的响应模型,包含所有字段
class ArchiveResponse(ArchiveCreate):
id: int
created_at: datetime
class Config:
orm_mode = True 允许 Pydantic 模型直接读取 ORM 对象
orm_mode = True
```

现在我们将所有的组件整合起来,编写 API 的核心逻辑。我们将实现以下接口:
在项目根目录下创建 main.py 文件,并写入以下完整代码。请注意代码中的依赖注入部分,这是 FastAPI 管理数据库会话的标准方式:
```python
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List, Optional
import models
import schemas
from database import engine, SessionLocal
创建数据库表(如果不存在)
models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="档案教育服务平台 API", version="1.0.0")
依赖注入函数:获取数据库会话
使用 yield 确保在请求处理完毕后自动关闭会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/archives/", response_model=schemas.ArchiveResponse, status_code=status.HTTP_201_CREATED)
def create_archive(archive: schemas.ArchiveCreate, db: Session = Depends(get_db)):
"""
创建新的教育档案
"""
将 Pydantic 模型转换为 ORM 模型实例
db_archive = models.EducationArchive(archive.dict())
db.add(db_archive)
db.commit()
db.refresh(db_archive)
return db_archive
@app.get("/archives/", response_model=List[schemas.ArchiveResponse])
def read_archives(category: Optional[str] = None, db: Session = Depends(get_db)):
"""
获取档案列表,可通过 category 参数进行筛选
"""
query = db.query(models.EducationArchive)
if category:
query = query.filter(models.EducationArchive.category == category)
return query.all()
@app.get("/archives/{archive_id}", response_model=schemas.ArchiveResponse)
def read_archive(archive_id: int, db: Session = Depends(get_db)):
"""
根据 ID 获取单条档案详情
"""
db_archive = db.query(models.EducationArchive).filter(models.EducationArchive.id == archive_id).first()
if db_archive is None:
raise HTTPException(status_code=404, detail="档案未找到")
return db_archive
@app.delete("/archives/{archive_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_archive(archive_id: int, db: Session = Depends(get_db)):
"""
根据 ID 删除档案
"""
db_archive = db.query(models.EducationArchive).filter(models.EducationArchive.id == archive_id).first()
if db_archive is None:
raise HTTPException(status_code=404, detail="档案未找到")
db.delete(db_archive)
db.commit()
return None
```
代码编写完成后,我们即可启动服务。FastAPI 推荐使用 Uvicorn 作为服务器。在项目根目录下,执行以下命令启动服务:
uvicorn main:app --reload
参数解释:main:app 指的是 main.py 文件中的 app 实例;--reload 参数开启了热重载模式,修改代码后服务会自动重启,非常适合开发调试。
看到终端输出 Application startup complete 且提示 Uvicorn running on http://127.0.0.1:8000,即表示服务启动成功。
FastAPI 自带自动生成的交互式 API 文档(Swagger UI),这是测试接口的神器。直接在浏览器中访问:
http://127.0.0.1:8000/docs
你将看到一个可视化的接口列表。以下是具体的测试步骤:
1. 测试新增档案(POST)
POST /archives/ 接口。{
"title": "Python 高级编程实战",
"instructor": "张三",
"category": "编程开发",
"description": "深入讲解 Python 异步编程与元类。",
"file_url": "http://example.com/python_advanced.mp4"
}
201 并在 Response body 中看到生成的数据(包含 id 和 created_at),说明接口正常。2. 测试获取列表(GET)
GET /archives/ 接口。category 输入框中输入“编程开发”进行筛选测试。3. 命令行测试(可选)
如果你习惯使用命令行工具,可以使用 curl 进行快速测试。例如,获取所有档案:
curl http://127.0.0.1:8000/archives/
或者新增一条档案:
curl -X POST "http://127.0.0.1:8000/archives/" \
-H "Content-Type: application/json" \
-d '{"title": "历史档案研究", "instructor": "李四", "category": "人文历史"}'
在实操过程中,可能会遇到以下常见问题,请参照解决方案:
1. 端口被占用错误
错误信息通常为:[Errno 48] Address already in use。这表示 8000 端口已被其他程序使用。你可以通过更换端口启动解决:
uvicorn main:app --port 8001
2. 模块导入错误
如果在启动时提示 ModuleNotFoundError,请确保你已经在虚拟环境中,并且所有依赖都已正确安装。可以重新运行 pip install -r requirements.txt。
3. 数据库锁定
SQLite 在高并发写入时可能会出现锁定。如果在测试中发现写入报错,请确保没有在数据库可视化工具(如 DB Browser for SQLite)中同时打开并锁定了 arch_edu.db 文件。
至此,一个基于 FastAPI 的档案教育服务平台后端原型已经搭建完成。你可以基于此基础继续扩展用户认证、文件上传下载等功能,逐步完善整个系统。