一、开发环境准备与依赖安装
在开始构建档案监督机构管理系统之前,必须先准备好标准的Python开发环境。本系统基于Python 3.9+版本开发,利用FastAPI框架提供高性能API接口,使用SQLAlchemy进行数据库ORM操作,SQLite作为本地数据库以实现零配置落地。
请在终端中依次执行以下命令,创建项目目录并安装核心依赖包:
- 创建项目目录:在本地磁盘新建一个文件夹,命名为
archive_supervision_system,并进入该目录。
- 安装核心库:执行以下命令安装FastAPI、Uvicorn服务器、SQLAlchemy数据库工具包:
```bash
pip install fastapi uvicorn sqlalchemy
```
二、数据库模型设计与配置
为了体现“档案制度建设”与“监督机构健全”的核心需求,我们需要设计两张核心数据表:supervision_institutions(监督机构表)和 archive_systems(档案制度表)。同时,为了保证数据的可追溯性,还需要一张 audit_logs(审计日志表)。
在项目根目录下创建 database.py 文件,并写入以下完整的数据库配置与模型代码。该代码定义了数据库连接、表结构及字段关系,可直接复制使用。
```python
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime
数据库配置:使用SQLite本地文件,无需安装数据库服务
SQLALCHEMY_DATABASE_URL = "sqlite:///./archive_supervision.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
监督机构表:对应“监督机构健全”
class SupervisionInstitution(Base):
__tablename__ = "supervision_institutions"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False, comment="机构名称")
level = Column(String(50), nullable=False, comment="机构级别(如:国家级、省级)")
responsibility = Column(Text, comment="监督职责描述")
contact_info = Column(String(200), comment="联系方式")
created_at = Column(DateTime, default=datetime.now)
档案制度表:对应“档案制度建设”
class ArchiveSystem(Base):
__tablename__ = "archive_systems"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(200), nullable=False, comment="制度标题")
content = Column(Text, nullable=False, comment="制度详细内容")
institution_id = Column(Integer, ForeignKey("supervision_institutions.id"), comment="关联的监督机构ID")
version = Column(String(20), default="V1.0", comment="制度版本号")
status = Column(String(20), default="Draft", comment="状态:Draft/Active/Archived")
created_at = Column(DateTime, default=datetime.now)
关联关系:一个制度属于一个机构
institution = relationship("SupervisionInstitution", backref="archive_systems")
审计日志表:记录所有操作行为
class AuditLog(Base):
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, index=True)
action = Column(String(50), nullable=False, comment="操作类型")
target = Column(String(100), comment="操作对象")
details = Column(Text, comment="操作详情")
timestamp = Column(DateTime, default=datetime.now)
初始化数据库表
def init_db():
Base.metadata.create_all(bind=engine)
```
三、后端核心业务逻辑实现
接下来编写核心业务逻辑,包括机构的录入、制度的发布以及审计日志的记录。创建 main.py 文件,并填入以下代码。这段代码实现了完整的CRUD接口,并包含数据校验逻辑。
```python
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import Optional, List
from database import SessionLocal, init_db, SupervisionInstitution, ArchiveSystem, AuditLog
初始化应用与数据库
app = FastAPI(title="档案监督机构管理系统")
init_db()
依赖项:获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Pydantic模型:用于数据验证
class InstitutionCreate(BaseModel):
name: str
level: str
responsibility: str
contact_info: str
class ArchiveSystemCreate(BaseModel):
title: str
content: str
institution_id: int
version: str = "V1.0"
API 1: 创建监督机构
@app.post("/api/institutions/")
def create_institution(inst: InstitutionCreate, db: Session = Depends(get_db)):
检查机构是否已存在
existing = db.query(SupervisionInstitution).filter(SupervisionInstitution.name == inst.name).first()
if existing:
raise HTTPException(status_code=400, detail="该机构名称已存在")
db_inst = SupervisionInstitution(inst.dict())
db.add(db_inst)
记录审计日志
log = AuditLog(action="CREATE_INSTITUTION", target=inst.name, details=f"创建新监督机构: {inst.name}")
db.add(log)
db.commit()
db.refresh(db_inst)
return {"code": 200, "message": "机构创建成功", "data": db_inst}
API 2: 发布档案制度
@app.post("/api/systems/")
def create_archive_system(sys: ArchiveSystemCreate, db: Session = Depends(get_db)):
校验关联机构是否存在(确保监督机构健全)
institution = db.query(SupervisionInstitution).filter(SupervisionInstitution.id == sys.institution_id).first()
if not institution:
raise HTTPException(status_code=404, detail="关联的监督机构不存在,无法发布制度")
db_sys = ArchiveSystem(sys.dict(), status="Active")
db.add(db_sys)
记录审计日志
log = AuditLog(action="PUBLISH_SYSTEM", target=sys.title, details=f"机构ID {sys.institution_id} 发布了新制度: {sys.title}")
db.add(log)
db.commit()
db.refresh(db_sys)
return {"code": 200, "message": "制度发布成功", "data": db_sys}
API 3: 查询所有监督机构及其制度
@app.get("/api/institutions/")
def list_institutions(db: Session = Depends(get_db)):
institutions = db.query(SupervisionInstitution).all()
result = []
for inst in institutions:
result.append({
"id": inst.id,
"name": inst.name,
"level": inst.level,
"systems_count": len(inst.archive_systems)
})
return {"code": 200, "data": result}
API 4: 查询审计日志
@app.get("/api/logs/")
def list_logs(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
logs = db.query(AuditLog).order_by(AuditLog.timestamp.desc()).offset(skip).limit(limit).all()
return {"code": 200, "data": logs}
```
四、系统启动与功能验证
代码编写完成后,通过命令行启动服务,并使用cURL命令进行功能测试,确保“监督机构健全”和“制度建设”的逻辑闭环。
1. 启动服务

在终端执行以下命令启动FastAPI服务器:
```bash
uvicorn main:app --reload --host 0.0.0.0 --port 8000
``>
当终端显示 Application startup complete 时,说明服务已成功启动,并自动在本地生成了 archive_supervision.db 数据库文件。
2. 创建监督机构(健全机构)
首先创建一个监督机构,模拟“机构健全”的过程。打开新的终端窗口,执行以下命令:
```bash
curl -X POST "http://127.0.0.1:8000/api/institutions/" \
-H "Content-Type: application/json" \
-d '{
"name": "省档案局监督处",
"level": "省级",
"responsibility": "负责全省档案制度建设的监督与指导",
"contact_info": "020-12345678"
}'
``>
预期返回结果:系统应返回状态码200及创建的机构信息,包含自动生成的ID。
3. 发布档案制度(制度建设)
接着,基于刚创建的机构ID(假设为1),发布一项档案制度。执行以下命令:
```bash
curl -X POST "http://127.0.0.1:8000/api/systems/" \
-H "Content-Type: application/json" \
-d '{
"title": "电子档案归档管理办法",
"content": "本办法规定了电子档案的归档范围、流程及存储标准...",
"institution_id": 1,
"version": "V2.0"
}'
``>
预期返回结果:返回制度发布成功信息。如果 institution_id 不存在,系统将返回404错误,从而强制要求必须先有健全的机构才能发布制度。
4. 查询监督与审计数据
查询机构列表和审计日志,验证数据的一致性:
```bash
curl "http://127.0.0.1:8000/api/institutions/"
curl "http://127.0.0.1:8000/api/logs/"
``>
通过以上步骤,我们完成了一个轻量级但功能闭环的档案监督机构管理系统。该系统通过数据库外键约束强制了“监督机构健全”与“档案制度建设”的关联关系,并通过审计日志实现了全流程监督,所有代码均无需额外配置即可运行。