DevNull 知识库

Appearance

52｜数据模型与数据库设计 ​

需求确定后，第一步是设计数据模型。资产、任务、用户、事件这四类对象需要在数据库中持久化，它们之间的关系决定了表结构和外键设计。SQLAlchemy 模型定义后，用 Alembic 生成迁移脚本，把结构同步到数据库。

一、表结构设计 ​

users（用户） ​

python
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    username = Column(String(32), nullable=False, unique=True)
    password_hash = Column(String(128), nullable=False)
    role = Column(String(20), default="user")   # admin / user
    created_at = Column(DateTime, default=datetime.utcnow)

assets（资产/服务器） ​

python
class Asset(Base):
    __tablename__ = "assets"

    id = Column(Integer, primary_key=True)
    hostname = Column(String(64), nullable=False)
    ip = Column(String(15), nullable=False)
    status = Column(String(20), default="running")
    env = Column(String(20), default="dev")     # dev / test / prod
    owner_id = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    owner = relationship("User", back_populates="assets")

tasks（任务） ​

python
class Task(Base):
    __tablename__ = "tasks"

    id = Column(Integer, primary_key=True)
    asset_id = Column(Integer, ForeignKey("assets.id"), nullable=False)
    action = Column(String(50), nullable=False)   # restart / check / deploy
    status = Column(String(20), default="pending")  # pending / running / done / failed
    result = Column(Text)
    created_by = Column(Integer, ForeignKey("users.id"))
    created_at = Column(DateTime, default=datetime.utcnow)
    completed_at = Column(DateTime)

    asset = relationship("Asset", back_populates="tasks")
    creator = relationship("User")

events（事件/审计日志） ​

python
class Event(Base):
    __tablename__ = "events"

    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    action = Column(String(50), nullable=False)   # create_asset / delete_asset
    target_type = Column(String(50))              # asset / task / user
    target_id = Column(Integer)
    detail = Column(Text)
    created_at = Column(DateTime, default=datetime.utcnow)

    user = relationship("User")

二、Pydantic 模型 ​

请求和响应用 Pydantic 模型约束：

python
from pydantic import BaseModel, Field
from datetime import datetime

# Asset
class AssetBase(BaseModel):
    hostname: str = Field(..., min_length=1, max_length=64)
    ip: str = Field(..., regex=r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
    env: str = "dev"

class AssetCreate(AssetBase):
    pass

class AssetUpdate(BaseModel):
    hostname: str | None = Field(None, min_length=1, max_length=64)
    ip: str | None = Field(None, regex=r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
    status: str | None = None
    env: str | None = None

class AssetOut(AssetBase):
    id: int
    status: str
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True   # 允许从 ORM 对象创建

三、Alembic 迁移 ​

bash
cd backend
uv run alembic revision --autogenerate -m "init tables"
uv run alembic upgrade head

生成的迁移脚本检查点：

python
def upgrade():
    op.create_table("users", ...)
    op.create_table("assets", ...)
    op.create_table("tasks", ...)
    op.create_table("events", ...)

def downgrade():
    op.drop_table("events")
    op.drop_table("tasks")
    op.drop_table("assets")
    op.drop_table("users")

四、种子数据 ​

开发环境需要初始用户：

python
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def seed_data():
    admin = User(
        username="admin",
        password_hash=pwd_context.hash("admin123"),
        role="admin",
    )
    db.add(admin)
    db.commit()

五、常见错误 ​

外键类型不匹配 ​

python
# 错误：users.id 是 Integer，assets.owner_id 是 String
owner_id = Column(String, ForeignKey("users.id"))

# 正确
owner_id = Column(Integer, ForeignKey("users.id"))

忘记 onupdate ​

python
# 错误：updated_at 只在创建时设置，后续修改不变
updated_at = Column(DateTime, default=datetime.utcnow)

# 正确
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

响应模型缺少 from_attributes ​

python
# 错误：SQLAlchemy 对象不能直接转成 Pydantic 模型
class AssetOut(AssetBase):
    id: int
    # 缺少 Config.from_attributes

# 正确
class AssetOut(AssetBase):
    id: int
    class Config:
        from_attributes = True