DevNull 知识库

Appearance

后端接口实现 ​

22 篇定义了四张表的结构，接下来把它们变成 API——前端能调用的接口。重点在资产的增删改查（CRUD），任务和事件的接口结构类似。

一、Pydantic Schema ​

先定义 API 的数据格式——请求体长什么样、响应体长什么样，跟数据库模型分开：

python
# backend/app/schemas.py
from pydantic import BaseModel
from datetime import datetime


# ---- 资产 ----
class AssetCreate(BaseModel):
    hostname: str
    ip: str
    port: int = 22
    role: str
    env: str
    remark: str | None = None


class AssetUpdate(BaseModel):
    hostname: str | None = None
    ip: str | None = None
    port: int | None = None
    role: str | None = None
    env: str | None = None
    status: str | None = None
    remark: str | None = None


class AssetResponse(BaseModel):
    id: int
    hostname: str
    ip: str
    port: int
    role: str
    env: str
    status: str
    remark: str | None
    created_at: datetime

    model_config = {"from_attributes": True}  # 从 SQLAlchemy 模型实例自动取值

Schema 和 Model 分开的原因：Model 描述数据库存储，Schema 描述 API 传输。创建时 hashed_password 不出现在请求里（用户传明文密码，后端哈希后存 Model），列表响应时 content 这类大字段不返回——这些差异通过分两套类来控制。

model_config = {"from_attributes": True} 让 Pydantic 从 SQLAlchemy 模型实例自动取属性值——AssetResponse.model_validate(asset_model) 直接转，不用手动一个个字段赋值。

二、数据库会话和认证依赖 ​

复用前面基础篇的依赖：

python
# backend/app/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from app.config import settings

engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)


def get_db():
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()
python
# backend/app/auth.py（简化版，完整版见 14 篇）
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/users/login")


async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    # 解析 Token、查用户（完整逻辑见 14 篇）
    ...


async def get_operator_user(user = Depends(get_current_user)):
    """operator 和 admin 都能执行操作，viewer 不行。"""
    if user.role == "viewer":
        raise HTTPException(status_code=403, detail="权限不足")
    return user

get_operator_user 在 get_current_user 基础上加权限检查——viewer 只读，operator 和 admin 能执行操作。

三、资产接口 ​

1 列表（分页 + 过滤） ​

python
# backend/app/routers/assets.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session

router = APIRouter()


@router.get("", response_model=list[AssetResponse])
def list_assets(
    env: str | None = None,           # 按环境过滤
    keyword: str | None = None,       # 按主机名或 IP 搜索
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user),  # 需要登录
):
    """列出资产，支持按环境过滤和关键词搜索。"""
    query = db.query(Asset)

    if env:
        query = query.filter(Asset.env == env)

    if keyword:
        query = query.filter(
            (Asset.hostname.contains(keyword)) | (Asset.ip.contains(keyword))
        )

    assets = query.offset(skip).limit(limit).order_by(Asset.id.desc()).all()
    return assets

Query(20, ge=1, le=100) 限制 limit 参数在 1-100 之间——防止前端传 limit=999999 拖垮数据库（15 篇讲过）。

current_user = Depends(get_current_user) 要求登录才能访问，但不检查角色——viewer 也能看资产列表。创建和删除操作才加 get_operator_user 检查写权限。

2 详情 ​

python
@router.get("/{asset_id}", response_model=AssetResponse)
def get_asset(asset_id: int, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
    asset = db.query(Asset).filter(Asset.id == asset_id).first()
    if not asset:
        raise HTTPException(status_code=404, detail="资产不存在")
    return asset

3 创建 ​

python
@router.post("", response_model=AssetResponse)
def create_asset(
    asset: AssetCreate,
    db: Session = Depends(get_db),
    user = Depends(get_operator_user),  # 需要写权限
):
    new_asset = Asset(**asset.model_dump())
    db.add(new_asset)
    db.commit()
    db.refresh(new_asset)

    # 记审计事件
    log_event(db, user_id=user.id, action="create_asset",
              target_type="asset", target_id=new_asset.id,
              detail=f"创建资产 {new_asset.hostname}")

    return new_asset

Asset(**asset.model_dump()) 拆开看：asset.model_dump() 把 Pydantic 对象转成字典（{"hostname": "web-01", "ip": "192.168.1.10", ...}），前面的 ** 把字典展开成关键字参数——等价于 Asset(hostname="web-01", ip="192.168.1.10", ...)。字段名对得上就直接赋值，不用一个个写。log_event 是审计日志的辅助函数，每个写操作都调一次。

4 更新和删除 ​

python
@router.patch("/{asset_id}", response_model=AssetResponse)
def update_asset(
    asset_id: int,
    update: AssetUpdate,
    db: Session = Depends(get_db),
    user = Depends(get_operator_user),
):
    asset = db.query(Asset).filter(Asset.id == asset_id).first()
    if not asset:
        raise HTTPException(status_code=404, detail="资产不存在")

    # 只更新传了的字段（11 篇的 exclude_unset 技巧）
    for field, value in update.model_dump(exclude_unset=True).items():
        setattr(asset, field, value)  # setattr(asset, "hostname", "web-02") 等价于 asset.hostname = "web-02"

    db.commit()
    db.refresh(asset)

    log_event(db, user_id=user.id, action="update_asset",
              target_type="asset", target_id=asset_id)

    return asset


@router.delete("/{asset_id}")
def delete_asset(asset_id: int, db: Session = Depends(get_db), user=Depends(get_operator_user)):
    asset = db.query(Asset).filter(Asset.id == asset_id).first()
    if not asset:
        raise HTTPException(status_code=404, detail="资产不存在")

    db.delete(asset)
    db.commit()

    log_event(db, user_id=user.id, action="delete_asset",
              target_type="asset", target_id=asset_id)

    return {"message": "已删除"}

四、审计事件辅助函数 ​

python
# backend/app/routers/events.py
def log_event(db: Session, user_id: int, action: str,
              target_type: str = None, target_id: int = None, detail: str = None):
    """记录审计事件。每个写操作调用一次。"""
    event = Event(
        user_id=user_id,
        action=action,
        target_type=target_type,
        target_id=target_id,
        detail=detail,
    )
    db.add(event)
    db.commit()


@router.get("", response_model=list[EventResponse])
def list_events(
    skip: int = 0,
    limit: int = Query(50, le=200),
    db: Session = Depends(get_db),
    user = Depends(get_current_user),
):
    """查询事件日志（只有 admin 能看全部）。"""
    if user.role != "admin":
        # 非 admin 只看自己的操作记录
        events = db.query(Event).filter(Event.user_id == user.id)
    else:
        events = db.query(Event)

    return events.order_by(Event.id.desc()).offset(skip).limit(limit).all()

log_event 在每个写操作（创建、更新、删除）里调用——审计日志跟业务操作在同一个事务里，要么一起成功，要么一起失败。事件查询接口加了权限控制：admin 能看所有人的操作记录，普通用户只看自己的。

五、验证接口 ​

启动后端，访问 http://localhost:8000/docs——FastAPI 自动生成的 Swagger 文档里能看到所有接口，还能直接在页面上测试：

bash
# 登录拿 Token
curl -X POST http://localhost:8000/api/users/login \
  -d 'username=admin&password=admin123'

# 带 Token 创建资产
curl -X POST http://localhost:8000/api/assets \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{"hostname": "web-01", "ip": "192.168.1.10", "role": "web", "env": "prod"}'

# 列出资产
curl http://localhost:8000/api/assets \
  -H "Authorization: Bearer <token>"

接口跑通后，前端就能接上了——下一篇写前端页面。