Files

4.1 KiB

请完成以下任务。项目在 /home/enzo/projects/hive-manage-v3/,已有 backend/ 目录和 venv (.venv/)。

步骤 1: 初始化 Alembic

在 backend/ 目录执行:

cd backend && .venv/bin/alembic init alembic

步骤 2: 修改 backend/alembic/env.py

配置异步引擎,关键改动:

  • 导入 from app.database import Base, engine
  • 导入 from app.models import * # noqa: F401,F403
  • target_metadata = Base.metadata
  • 重写 run_migrations_online 以使用异步引擎:
from app.config import settings
from app.database import Base, engine
from app.models import *  # noqa: F401,F403

target_metadata = Base.metadata

def run_migrations_offline():
    url = settings.database_url
    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
    with context.begin_transaction():
        context.run_migrations()

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    async with engine.begin() as conn:
        await conn.run_sync(do_run_migrations)

def run_migrations_online():
    import asyncio
    asyncio.run(run_async_migrations())

步骤 3: 创建 backend/app/models/__init__.py

空的 __init__.py。

步骤 4: 创建 backend/app/models/user.py

"""用户 ORM 模型。"""
import uuid
from datetime import datetime
from sqlalchemy import String, Boolean, DateTime, Enum as SAEnum
from sqlalchemy.orm import Mapped, mapped_column
import enum

from app.database import Base


class UserRole(str, enum.Enum):
    OWNER = "owner"
    OPERATOR = "operator"
    VIEWER = "viewer"


class User(Base):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[UserRole] = mapped_column(SAEnum(UserRole), default=UserRole.VIEWER, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    is_approved: Mapped[bool] = mapped_column(Boolean, default=False)
    login_attempts: Mapped[int] = mapped_column(default=0)
    locked_until: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

步骤 5: 创建 backend/app/models/audit.py

"""审计日志 ORM 模型。"""
import uuid
from datetime import datetime
from sqlalchemy import String, DateTime, Text
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class AuditLog(Base):
    __tablename__ = "audit_logs"

    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    user_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
    username: Mapped[str] = mapped_column(String(50), nullable=False)
    action: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
    target: Mapped[str] = mapped_column(String(255), nullable=False)
    detail: Mapped[str | None] = mapped_column(Text, nullable=True)
    ip_address: Mapped[str | None] = mapped_column(String(45), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

步骤 6: 生成并执行迁移

cd backend && .venv/bin/alembic revision --autogenerate -m "create users and audit_logs tables"
cd backend && .venv/bin/alembic upgrade head

步骤 7: 验证并提交

cd /home/enzo/projects/hive-manage-v3 && git add backend/alembic/ backend/app/models/ && git commit -m "feat: Alembic setup, User and AuditLog models with migration"

请严格按照以上步骤执行。遇到 pip/alembic 命令时使用 backend/.venv/bin/ 前缀。