feat(system): 新增操作日志表与 V2 迁移
需求 2.5.1 要求记录用户管理操作，先落表结构： - operation_logs 表：冗余 username 保证账号删除后仍可追溯， user_id 外键 ON DELETE SET NULL，success 默认 '0' 与 login_logs 口径一致 - 截断常量（USERNAME/ERROR_MESSAGE/REQUEST_DATA）集中定义，供写入方引用 - 修复既有 V1 迁移与迁移模板的 downgrade：索引是外键唯一支撑索引， 先 drop_index 会报 errno 1553，改为只 drop_table - 同步 db/migration/V2 与 init-dev.sql，迁移版本号更新为 a45a8087831b Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxinandClaude Code committed
1 parent
273ebf4657
commit
ad5fe4a411
9 files changed
+365
-30
No files matched your search
@@ -5,7 +5,7 @@ Revises: ${down_revision | comma,n}
|
||||
Create Date: ${create_date}
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
from collections.abc import Sequence
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
@@ -13,16 +13,22 @@ ${imports if imports else ""}
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = ${repr(up_revision)}
|
||||
down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
|
||||
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
|
||||
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
|
||||
down_revision: str | Sequence[str] | None = ${repr(down_revision)}
|
||||
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
|
||||
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Upgrade schema."""
|
||||
"""应用迁移。"""
|
||||
${upgrades if upgrades else "pass"}
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
"""回滚迁移。
|
||||
|
||||
**注意（autogenerate 不会替你处理）**：生成的 `op.drop_index` 与
|
||||
`op.drop_table` 顺序需人工检查 —— 若某索引是外键的唯一支撑索引，
|
||||
单独 drop_index 会报 MySQL 1553（needed in a foreign key constraint）。
|
||||
**整表回滚应只保留 `op.drop_table`**，DROP TABLE 会连带删除索引与外键。
|
||||
"""
|
||||
${downgrades if downgrades else "pass"}
|
||||
@@ -337,17 +337,19 @@ def upgrade() -> None:
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.drop_index("idx_user_roles_role_id", table_name="user_roles")
|
||||
"""回滚迁移。
|
||||
|
||||
只保留 drop_table —— DROP TABLE 会连带删除索引与外键。
|
||||
原 autogenerate 输出里逐个 drop_index 会失败：idx_user_roles_role_id 与
|
||||
idx_role_permissions_permission_id 分别是外键的唯一支撑索引，
|
||||
MySQL 拒绝单独删除（errno 1553），回滚会中断在半途。
|
||||
|
||||
删除顺序遵循外键依赖：先子表（user_roles / role_permissions / login_logs），
|
||||
再父表（users / roles / permissions）。
|
||||
"""
|
||||
op.drop_table("user_roles")
|
||||
op.drop_index("idx_role_permissions_permission_id", table_name="role_permissions")
|
||||
op.drop_table("role_permissions")
|
||||
op.drop_index("idx_login_logs_username", table_name="login_logs")
|
||||
op.drop_index("idx_login_logs_create_time", table_name="login_logs")
|
||||
op.drop_table("login_logs")
|
||||
op.drop_index(op.f("idx_users_phone"), table_name="users")
|
||||
op.drop_table("users")
|
||||
op.drop_table("roles")
|
||||
op.drop_table("permissions")
|
||||
# ### end Alembic commands ###
|
||||
@@ -0,0 +1,109 @@
|
||||
"""新增操作日志表
|
||||
|
||||
Revision ID: a45a8087831b
|
||||
Revises: 6165691326d4
|
||||
Create Date: 2026-10-10 10:06:17.435697
|
||||
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects import mysql
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "a45a8087831b"
|
||||
down_revision: str | Sequence[str] | None = "6165691326d4"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""应用迁移。"""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.create_table(
|
||||
"operation_logs",
|
||||
sa.Column(
|
||||
"id", mysql.BIGINT(unsigned=True), autoincrement=True, nullable=False, comment="主键ID"
|
||||
),
|
||||
sa.Column(
|
||||
"user_id",
|
||||
mysql.BIGINT(unsigned=True),
|
||||
nullable=True,
|
||||
comment="操作人ID（账号删除后置空）",
|
||||
),
|
||||
sa.Column(
|
||||
"username",
|
||||
sa.String(length=64),
|
||||
nullable=False,
|
||||
comment="操作人用户名（冗余存储便于追溯）",
|
||||
),
|
||||
sa.Column("module", sa.String(length=64), nullable=False, comment="所属模块，如 用户管理"),
|
||||
sa.Column(
|
||||
"action", sa.String(length=32), nullable=False, comment="操作类型：新增/修改/删除/导出"
|
||||
),
|
||||
sa.Column(
|
||||
"summary", sa.String(length=255), nullable=False, comment="操作摘要（中文，供人阅读）"
|
||||
),
|
||||
sa.Column("method", sa.String(length=8), nullable=True, comment="HTTP 方法"),
|
||||
sa.Column("path", sa.String(length=255), nullable=True, comment="请求路径"),
|
||||
sa.Column("ip", sa.String(length=45), nullable=True, comment="客户端IP（45位兼容IPv6）"),
|
||||
sa.Column(
|
||||
"user_agent",
|
||||
sa.String(length=255),
|
||||
nullable=True,
|
||||
comment="客户端User-Agent（截断存储）",
|
||||
),
|
||||
sa.Column("request_data", sa.Text(), nullable=True, comment="请求参数（已脱敏）"),
|
||||
sa.Column(
|
||||
"success",
|
||||
mysql.TINYINT(),
|
||||
server_default="0",
|
||||
nullable=False,
|
||||
comment="是否成功：1成功 0失败",
|
||||
),
|
||||
sa.Column(
|
||||
"error_message",
|
||||
sa.String(length=500),
|
||||
nullable=True,
|
||||
comment="失败原因（写入方须截断）",
|
||||
),
|
||||
sa.Column("duration_ms", sa.Integer(), nullable=True, comment="处理耗时（毫秒）"),
|
||||
sa.Column(
|
||||
"create_time",
|
||||
sa.DateTime(),
|
||||
server_default=sa.text("CURRENT_TIMESTAMP"),
|
||||
nullable=False,
|
||||
comment="创建时间",
|
||||
),
|
||||
sa.ForeignKeyConstraint(
|
||||
["user_id"], ["users.id"], name=op.f("fk_operation_logs_user_id"), ondelete="SET NULL"
|
||||
),
|
||||
sa.PrimaryKeyConstraint("id", name=op.f("pk_operation_logs")),
|
||||
comment="操作日志表",
|
||||
mysql_charset="utf8mb4",
|
||||
mysql_collate="utf8mb4_general_ci",
|
||||
mysql_engine="InnoDB",
|
||||
)
|
||||
op.create_index(
|
||||
"idx_operation_logs_create_time", "operation_logs", ["create_time"], unique=False
|
||||
)
|
||||
op.create_index("idx_operation_logs_module", "operation_logs", ["module"], unique=False)
|
||||
op.create_index("idx_operation_logs_user_id", "operation_logs", ["user_id"], unique=False)
|
||||
op.create_index("idx_operation_logs_username", "operation_logs", ["username"], unique=False)
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""回滚迁移。
|
||||
|
||||
**注意（autogenerate 不会替你处理）**：生成的 `op.drop_index` 与
|
||||
`op.drop_table` 顺序需人工检查 —— 若某索引是外键的唯一支撑索引，
|
||||
单独 drop_index 会报 MySQL 1553（needed in a foreign key constraint）。
|
||||
**整表回滚应只保留 `op.drop_table`**，DROP TABLE 会连带删除索引与外键。
|
||||
"""
|
||||
# 整表回滚只保留 drop_table —— DROP TABLE 会连带删除索引与外键。
|
||||
# 逐个 drop_index 会失败：idx_operation_logs_user_id 是外键
|
||||
# fk_operation_logs_user_id 的唯一支撑索引，MySQL 拒绝删除（errno 1553）。
|
||||
op.drop_table("operation_logs")
|
||||
@@ -11,12 +11,14 @@
|
||||
|
||||
from app.models.base import Base, TimestampMixin
|
||||
from app.models.login_log import LoginLog
|
||||
from app.models.operation_log import OperationLog
|
||||
from app.models.role import Permission, Role, RolePermission
|
||||
from app.models.user import User, UserRole
|
||||
|
||||
__all__ = [
|
||||
"Base",
|
||||
"LoginLog",
|
||||
"OperationLog",
|
||||
"Permission",
|
||||
"Role",
|
||||
"RolePermission",
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""操作日志模型。
|
||||
|
||||
需求 2.5.1 要求「记录用户登录、操作行为及计提参数/系数表变更，支持日志查询」。
|
||||
登录行为已在 `login_logs` 中单独记录（登录失败也要留痕，且不依赖认证上下文）；
|
||||
本表记录**已认证用户的操作行为**，两者分工明确、互不重叠。
|
||||
|
||||
不继承 TimestampMixin：日志是追加写、不可变的，没有"更新时间"语义。
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text, text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from app.models.base import BIGINT_UNSIGNED, TINYINT_TYPE, Base, mysql_table_args
|
||||
|
||||
# 各文本列的上限。**写入方必须按此截断**，否则 MySQL 严格模式抛 1406
|
||||
# （Data too long），而按"审计失败不得影响业务"的约定该异常会被吞掉 ——
|
||||
# 结果是整条日志静默丢失。截断值集中定义于此，供 repository 层引用。
|
||||
USERNAME_MAX_LEN = 64
|
||||
MODULE_MAX_LEN = 64
|
||||
ACTION_MAX_LEN = 32
|
||||
SUMMARY_MAX_LEN = 255
|
||||
PATH_MAX_LEN = 255
|
||||
IP_MAX_LEN = 45
|
||||
USER_AGENT_MAX_LEN = 255
|
||||
ERROR_MESSAGE_MAX_LEN = 500
|
||||
# request_data 是 MySQL TEXT（上限 65535 字节）。脱敏后的 JSON 可能超限
|
||||
# （如批量操作提交数千个 ID），超限同样会抛 1406，故写入方需按字节截断。
|
||||
REQUEST_DATA_MAX_BYTES = 60000
|
||||
|
||||
|
||||
class OperationLog(Base):
|
||||
"""操作日志表。
|
||||
|
||||
记录原则与 `login_logs` 一致：
|
||||
- **请求参数必须脱敏后入库**（复用 `core.logging.scrub`），日志是最容易泄漏
|
||||
密码与令牌的地方
|
||||
- 用户名冗余存储而非只存 user_id —— 账号被删除后仍能追溯操作者，
|
||||
故 `username` 单独建索引（与 `login_logs` 保持一致）
|
||||
- 写入失败不得影响业务（审计可丢，业务不可断）
|
||||
"""
|
||||
|
||||
__tablename__ = "operation_logs"
|
||||
__table_args__ = mysql_table_args(
|
||||
"操作日志表",
|
||||
Index("idx_operation_logs_user_id", "user_id"),
|
||||
# 账号删除后 user_id 被置空，只能按用户名追溯，故该列必须有索引，
|
||||
# 否则日志表增长后查询退化为全表扫描
|
||||
Index("idx_operation_logs_username", "username"),
|
||||
Index("idx_operation_logs_module", "module"),
|
||||
Index("idx_operation_logs_create_time", "create_time"),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(
|
||||
BIGINT_UNSIGNED, primary_key=True, autoincrement=True, comment="主键ID"
|
||||
)
|
||||
user_id: Mapped[int | None] = mapped_column(
|
||||
BIGINT_UNSIGNED,
|
||||
ForeignKey("users.id", ondelete="SET NULL"),
|
||||
comment="操作人ID（账号删除后置空）",
|
||||
)
|
||||
username: Mapped[str] = mapped_column(
|
||||
String(USERNAME_MAX_LEN), nullable=False, comment="操作人用户名（冗余存储便于追溯）"
|
||||
)
|
||||
module: Mapped[str] = mapped_column(
|
||||
String(MODULE_MAX_LEN), nullable=False, comment="所属模块，如 用户管理"
|
||||
)
|
||||
action: Mapped[str] = mapped_column(
|
||||
String(ACTION_MAX_LEN), nullable=False, comment="操作类型：新增/修改/删除/导出"
|
||||
)
|
||||
summary: Mapped[str] = mapped_column(
|
||||
String(SUMMARY_MAX_LEN), nullable=False, comment="操作摘要（中文，供人阅读）"
|
||||
)
|
||||
|
||||
method: Mapped[str | None] = mapped_column(String(8), comment="HTTP 方法")
|
||||
path: Mapped[str | None] = mapped_column(String(PATH_MAX_LEN), comment="请求路径")
|
||||
ip: Mapped[str | None] = mapped_column(String(IP_MAX_LEN), comment="客户端IP（45位兼容IPv6）")
|
||||
user_agent: Mapped[str | None] = mapped_column(
|
||||
String(USER_AGENT_MAX_LEN), comment="客户端User-Agent（截断存储）"
|
||||
)
|
||||
# 请求参数，**入库前已脱敏**；长文本故用 TEXT
|
||||
request_data: Mapped[str | None] = mapped_column(Text, comment="请求参数（已脱敏）")
|
||||
|
||||
# 默认 0 而非 1：与 login_logs 保持一致，且方向更保守 ——
|
||||
# 写入方漏传时记为「失败」，管理员会去核查；反之漏传记为「成功」则是
|
||||
# 审计漏报，且无声无息。写入方应显式传值（见 repository 层）。
|
||||
success: Mapped[int] = mapped_column(
|
||||
TINYINT_TYPE, nullable=False, server_default="0", comment="是否成功：1成功 0失败"
|
||||
)
|
||||
error_message: Mapped[str | None] = mapped_column(
|
||||
String(ERROR_MESSAGE_MAX_LEN), comment="失败原因（写入方须截断）"
|
||||
)
|
||||
duration_ms: Mapped[int | None] = mapped_column(Integer, comment="处理耗时（毫秒）")
|
||||
|
||||
create_time: Mapped[datetime] = mapped_column(
|
||||
DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment="创建时间"
|
||||
)
|
||||
@@ -1,13 +1,16 @@
|
||||
"""DDL 漂移检测。
|
||||
|
||||
结构定义同时存在于两处：
|
||||
- `db/migration/V1__*.sql` —— 事实来源，生产迁移用
|
||||
- `db/migration/V*.sql` —— 事实来源，生产迁移用
|
||||
- `db/init/init-dev.sql` —— 开发环境重置脚本，Docker 首次启动唯一会自动执行的位置
|
||||
|
||||
两份 DDL 手工同步迟早会漂移，且后果隐蔽：开发库的表结构与迁移脚本不一致，
|
||||
新人克隆下来建出的库和生产不一样，等到上线才发现。
|
||||
本测试把「靠自觉」变成「可自动拦截」，成本极低。
|
||||
|
||||
**必须比对全部 V*.sql 的合集**：早期版本只解析 V1，新增 V2 之后，
|
||||
两份 DDL 同时缺表仍会被判为「一致」，漂移守卫形同虚设。
|
||||
|
||||
注：本文件不需要容器，放在 integration 下是因为它校验的是交付产物而非应用代码。
|
||||
"""
|
||||
|
||||
@@ -17,8 +20,9 @@ from pathlib import Path
|
||||
import pytest
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
||||
MIGRATION_FILE = PROJECT_ROOT / "db" / "migration" / "V1__init_rbac_and_login_logs.sql"
|
||||
MIGRATION_DIR = PROJECT_ROOT / "db" / "migration"
|
||||
INIT_DEV_FILE = PROJECT_ROOT / "db" / "init" / "init-dev.sql"
|
||||
ALEMBIC_VERSIONS_DIR = PROJECT_ROOT / "backend" / "alembic" / "versions"
|
||||
|
||||
_CREATE_TABLE = re.compile(r"CREATE TABLE\s+`?(\w+)`?\s*\((.*?)\n\)", re.DOTALL)
|
||||
_COLUMN_LINE = re.compile(r"^\s{4}`?(\w+)`?\s", re.MULTILINE)
|
||||
@@ -38,9 +42,38 @@ def _parse_tables(sql: str) -> dict[str, set[str]]:
|
||||
return tables
|
||||
|
||||
|
||||
def _alembic_head() -> str:
|
||||
"""从迁移脚本推断当前 head 版本号。
|
||||
|
||||
纯文本解析而非调用 alembic 运行时 —— 后者需要连数据库，会让本测试
|
||||
从「秒级、无需容器」变成「必须起库」。
|
||||
head 的定义：没有被任何其他迁移的 down_revision 引用过的那个 revision。
|
||||
"""
|
||||
revisions: set[str] = set()
|
||||
referenced: set[str] = set()
|
||||
for path in ALEMBIC_VERSIONS_DIR.glob("*.py"):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
if found := re.search(r"^revision: str = ['\"](\w+)['\"]", text, re.MULTILINE):
|
||||
revisions.add(found.group(1))
|
||||
if found := re.search(r"^down_revision:.*?= ['\"](\w+)['\"]", text, re.MULTILINE):
|
||||
referenced.add(found.group(1))
|
||||
|
||||
heads = revisions - referenced
|
||||
assert len(heads) == 1, f"迁移链存在分叉或缺失，heads={sorted(heads)}"
|
||||
return heads.pop()
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def migration_tables() -> dict[str, set[str]]:
|
||||
return _parse_tables(MIGRATION_FILE.read_text(encoding="utf-8"))
|
||||
"""合并解析 db/migration/ 下全部 V*.sql。"""
|
||||
files = sorted(MIGRATION_DIR.glob("V*.sql"))
|
||||
assert files, f"未找到迁移脚本：{MIGRATION_DIR}"
|
||||
|
||||
merged: dict[str, set[str]] = {}
|
||||
for path in files:
|
||||
for table, columns in _parse_tables(path.read_text(encoding="utf-8")).items():
|
||||
merged.setdefault(table, set()).update(columns)
|
||||
return merged
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
@@ -52,7 +85,7 @@ class TestSchemaSync:
|
||||
"""两份 DDL 的结构一致性。"""
|
||||
|
||||
def test_文件存在(self) -> None:
|
||||
assert MIGRATION_FILE.is_file(), f"缺少迁移脚本：{MIGRATION_FILE}"
|
||||
assert list(MIGRATION_DIR.glob("V*.sql")), f"缺少迁移脚本：{MIGRATION_DIR}"
|
||||
assert INIT_DEV_FILE.is_file(), f"缺少开发初始化脚本：{INIT_DEV_FILE}"
|
||||
|
||||
def test_表集合一致(self, migration_tables: dict, init_dev_tables: dict) -> None:
|
||||
@@ -74,16 +107,28 @@ class TestSchemaSync:
|
||||
|
||||
|
||||
class TestSeedData:
|
||||
"""字典数据在两处应保持同步。"""
|
||||
"""字典数据与版本标记。"""
|
||||
|
||||
@pytest.mark.parametrize("role_code", ["admin", "user"])
|
||||
def test_角色初始化数据在两处都存在(self, role_code: str) -> None:
|
||||
migration_sql = MIGRATION_FILE.read_text(encoding="utf-8")
|
||||
migration_sql = "\n".join(
|
||||
path.read_text(encoding="utf-8") for path in sorted(MIGRATION_DIR.glob("V*.sql"))
|
||||
)
|
||||
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
|
||||
assert f"'{role_code}'" in migration_sql
|
||||
assert f"'{role_code}'" in init_sql
|
||||
|
||||
def test_开发脚本标记了迁移版本(self) -> None:
|
||||
"""不写 alembic_version 的话，在开发库上执行 upgrade head 会重复建表而失败。"""
|
||||
def test_开发脚本标记的版本号等于当前head(self) -> None:
|
||||
"""版本号必须等于 head，且不能只断言「存在」。
|
||||
|
||||
落后的后果：开发库重建后执行 `alembic upgrade head`，alembic 认为该迁移
|
||||
尚未应用，重复建表并报 1050 Table already exists，只能手工补版本或 down -v。
|
||||
"""
|
||||
init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
|
||||
assert "INSERT INTO alembic_version" in init_sql
|
||||
found = re.search(
|
||||
r"INSERT INTO alembic_version \(version_num\) VALUES \('(\w+)'\)", init_sql
|
||||
)
|
||||
assert found, "init-dev.sql 未写入 alembic_version"
|
||||
assert found.group(1) == _alembic_head(), (
|
||||
f"init-dev.sql 标记的版本号 {found.group(1)!r} 落后于当前 head {_alembic_head()!r}"
|
||||
)
|
||||
@@ -11,7 +11,15 @@ from sqlalchemy.dialects import mysql
|
||||
|
||||
from app.models import Base
|
||||
|
||||
EXPECTED_TABLES = {"users", "roles", "user_roles", "permissions", "role_permissions", "login_logs"}
|
||||
EXPECTED_TABLES = {
|
||||
"users",
|
||||
"roles",
|
||||
"user_roles",
|
||||
"permissions",
|
||||
"role_permissions",
|
||||
"login_logs",
|
||||
"operation_logs",
|
||||
}
|
||||
|
||||
|
||||
def _ddl(table_name: str) -> str:
|
||||
@@ -101,7 +109,7 @@ class TestTimestampColumns:
|
||||
def test_带时间戳的表含ON_UPDATE子句(self, table_name: str) -> None:
|
||||
assert "ON UPDATE CURRENT_TIMESTAMP" in _ddl(table_name)
|
||||
|
||||
@pytest.mark.parametrize("table_name", ["login_logs"])
|
||||
@pytest.mark.parametrize("table_name", ["login_logs", "operation_logs"])
|
||||
def test_仅追加的日志表不含ON_UPDATE子句(self, table_name: str) -> None:
|
||||
"""日志是不可变记录，没有"更新时间"语义。"""
|
||||
assert "ON UPDATE CURRENT_TIMESTAMP" not in _ddl(table_name)
|
||||
|
||||
Reference in new issue
Block a user