RomM 后端开发实战指南：FastAPI / SQLAlchemy 分层架构、认证作用域与 Alembic 迁移规范

RomM 后端开发实战指南：FastAPI / SQLAlchemy 分层架构、认证作用域与 Alembic 迁移规范

【免费下载链接】rommA beautiful, powerful, self-hosted ROM manager and player.项目地址: https://gitcode.com/GitHub_Trending/rom/romm

RomM 是一个自托管的 ROM 管理与游玩平台，其后端是基于 FastAPI 构建的 Python 服务（位于仓库backend/目录）。本指南以 RomM 官方后端开发技能文档（.claude/skills/backend-development/SKILL.md）为核心骨架，结合仓库源码逐层解析其分层架构、认证与作用域体系、Alembic 迁移卫生规范、OpenAPI→前端类型管线以及 uv/pytest/trunk 工作流。读完本文，你将能够在新功能开发时准确判断代码应落在哪个目录、如何保护路由、如何安全地编写数据库迁移，并跑通后端测试与代码质量检查。

技术栈与运行环境总览

RomM 后端技术选型非常集中，其核心依赖与职责如下：

完整架构参考文档位于 docs/BACKEND_ARCHITECTURE.md，其中包含目录地图、ER 图、全部 API 端点清单与认证流程。SKILL 文档明确要求：任何非平凡改动之前，先阅读该文档。

分层架构：代码应该放在哪里

SKILL 文档给出了一个清晰的目录地图，这也是 RomM 后端所有代码的组织原则：

endpoints/ FastAPI routers: request validation, response schemas, @protected_route scopes endpoints/responses/ Pydantic response schemas (these shape the OpenAPI → frontend types) endpoints/sockets/ Socket.IO event handlers handler/ Business logic, decoupled from HTTP ├ auth/ HybridAuthBackend (session/basic/bearer/OIDC/client-token), scopes, CSRF/session middleware ├ database/ Per-entity CRUD handlers (db_rom_handler, db_user_handler, …), engine/session factory ├ metadata/ One handler per provider; normalizes + ranks by priority └ filesystem/ ROM/asset/firmware file I/O, hashing, archive extraction adapters/services/ Typed external API clients (igdb.py + igdb_types.py, screenscraper.py, …) models/ SQLAlchemy ORM models (BaseModel adds created_at/updated_at) tasks/ RQ jobs — scheduled/ (cron) and manual/ (on-demand); base classes in tasks.py config/ Env-var loading (__init__.py) + YAML config manager (singleton) decorators/ @begin_session (DB session), @protected_route (auth + scopes) exceptions/ Custom exception hierarchy utils/ logger/ Shared helpers, structured logging alembic/ Migrations (env.py + versions/)

核心数据流是一条单向链路：

Endpoint → handler → (database | metadata | filesystem) → models/adapters

也就是说，endpoints/中的路由必须保持"薄"：只做请求校验、作用域（scope）检查、调用 handler、通过响应 schema 序列化输出。业务逻辑和裸 SQL 查询严禁写在 endpoint 中，这是 RomM 后端最根本的分层约束。

在 backend/main.py 中可以看到所有路由的挂载方式——20+ 个 router 统一以/api为前缀注册，Socket.IO 应用分别挂载在/ws与/netplay下，最后调用add_pagination(app)接入fastapi-pagination的分页能力。

中间件栈（执行顺序，由外到内）

backend/main.py 展示了一条完整的请求处理链，这也是分层架构在"横切关注点"上的体现：

Request → CORS → CSRF → Authentication → Session (Redis) → Context Vars → Endpoint Response ← CORS ← CSRF ← Authentication ← Session (Redis) ← Context Vars ← Endpoint

后端编码约定

SKILL 文档对新增代码有一组强制性约定，违反其中任何一条都会在代码评审中被要求修正：

• 命名：类PascalCase，函数/变量snake_case，常量UPPER_SNAKE_CASE，私有成员加_前缀。
• 数据库会话：handler 方法一律用@begin_session装饰，由装饰器负责注入并管理 SQLAlchemy 会话与事务，禁止随手手动开启会话。
• 异步：I/O 密集的 endpoint 和任务使用async/await；每次请求所需的httpx/aiohttp客户端从 context vars 获取（见 backend/utils/context.py），而不是每次调用新建客户端。
• 导入顺序：stdlib → 第三方 → 本地；禁止通配符导入；用TYPE_CHECKING块打破循环导入。
• 错误处理：优先抛出exceptions/中的自定义异常（如RomNotFoundInDatabaseException），而不是裸的HTTPException。
• 校验/SSRF 防护：文件系统使用前必须清洗文件名与路径，所有路径锚定在LIBRARY_BASE_PATH/RESOURCES_BASE_PATH/ASSETS_BASE_PATH配置根下。

@begin_session 的实现

backend/decorators/database.py 展示了@begin_session的真实实现：若调用方已传入session（说明处于既有的工作单元中）则直接复用；否则通过sync_session.begin()开启事务上下文，将session注入 kwargs 后调用原函数。SQLAlchemy 的ProgrammingError会被捕获并转为 500 响应。底层引擎与会话工厂定义在 backend/handler/database/base_handler.py：

sync_engine = create_engine( ConfigManager.get_db_engine(), pool_pre_ping=True, pool_recycle=DB_POOL_RECYCLE_SECONDS, echo=False, ) sync_session = sessionmaker(bind=sync_engine, expire_on_commit=False)

开启DEV_SQL_ECHO时还会通过 SQLAlchemy 事件钩子在before_cursor_execute/after_cursor_execute打印每条 SQL 与执行耗时，用于开发期排查。

BaseModel 的时间戳约定

所有 ORM 模型继承 backend/models/base.py 中的BaseModel，自动获得created_at与updated_at两个TIMESTAMP(timezone=True)列，均以 UTC 为准：

class BaseModel(DeclarativeBase): created_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), default=utc_now) updated_at: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True), default=utc_now, onupdate=utc_now)

该文件还定义了FILE_NAME_MAX_LENGTH=450、FILE_PATH_MAX_LENGTH=1000、FILE_EXTENSION_MAX_LENGTH=100等常量，以及从文件名派生no_tags/no_ext/extension列的辅助函数——这些派生列通过@validates钩子与源列保持同步。

认证与作用域（Auth & Scopes）

RomM 采用"角色 + 细粒度 scope"两级权限模型。

三级角色

角色定义在 backend/models/user.py 中：

• VIEWER：只读访问
• EDITOR：在 VIEWER 基础上可写 ROMs / platforms / assets
• ADMIN：在 EDITOR 基础上可管理用户、任务与日志

细粒度 scope

完整的 scope 枚举定义在 backend/handler/auth/constants.py，涵盖了me.read/write、roms.read/write、roms.user.read/write、platforms.*、assets.*、devices.*、firmware.*、collections.*、playlists.*、users.*、tasks.run、logs.read。scope 按层级累加组织：

READ_SCOPES: Final = list(READ_SCOPES_MAP.keys()) WRITE_SCOPES: Final = READ_SCOPES + list(WRITE_SCOPES_MAP.keys()) EDIT_SCOPES: Final = WRITE_SCOPES + list(EDIT_SCOPES_MAP.keys()) FULL_SCOPES: Final = EDIT_SCOPES + list(FULL_SCOPES_MAP.keys())

即：拥有写权限的角色天然继承所有只读 scope，ADMIN 进一步获得USERS_*、TASKS_RUN、LOGS_READ。同一文件还定义了 HS256 签名算法、默认 15 分钟的 OAuth token 有效期，以及会话 cookie 名romm_session。

@protected_route 保护路由

路由保护统一通过 backend/decorators/auth.py 中的protected_route完成：

@protected_route(router.get, "/", [Scope.ROMS_READ]) async def list_roms(request: Request, ...): ...

其内部做了三件事：用_requires_scopes包装函数（借助 starlette 的has_required_scope做 scope 守卫，并区分 401/403——未认证返回 401，已认证但 scope 不足返回 403）；将oauth2_password_bearer（tokenUrl 为/token）作为Security依赖注入；同时挂载HTTPBasic(auto_error=False)支持 Basic 认证。前端会镜像这些 scope，改动时必须保持两端对齐，否则会出现前端可调、后端拒绝或反之的权限错配。

新增功能的四类标准操作

SKILL 文档把最常见的开发任务归纳为四类，每类都有明确的落点：

1. 新增 Endpoint

• 在正确的endpoints/*router 中添加路由；
• 在endpoints/responses/添加 Pydantic 响应 schema；
• 用@protected_route强制 scope；
• 逻辑委托给 handler。
• 注意：如果响应形状（response shape）发生变化，前端必须重新生成类型（见下文"OpenAPI → 前端类型"一节）。

2. 模型 / schema 变更

• 修改models/下的 ORM 模型；
• 创建对应的 Alembic 迁移（见下一节）；
• 同步更新响应 schema，保证 OpenAPI 文档准确。

3. 新增元数据提供方

• 在adapters/services/<name>.py中编写带类型定义的 API 客户端，配套<name>_types.py类型文件；
• 在handler/metadata/<name>_handler.py中编写 handler，把各 provider 的异构数据归一化为通用形状，并接入优先级排序。

仓库中已有 IGDB、ScreenScraper、MobyGames、RetroAchievements、SteamGridDB、Hasheous、HLTB、Steam、TGDB、Flashpoint、gamelist.xml、Libretro 缩略图、PlayMatch、LaunchBox 等 15+ 个 provider handler，全部遵循"typed client + normalizing handler"的模式。

4. 新增后台任务

• 在tasks/scheduled/（周期任务）或tasks/manual/（按需任务）中继承Task/PeriodicTask基类；
• 在startup.py中注册定时任务。

任务基类定义在 backend/tasks/tasks.py：Task抽象类包含title、description、task_type、enabled、manual_run、cron_string、timeout等字段，子类只需实现async def run()。任务类型由TaskType枚举约束（SCAN、CONVERSION、CLEANUP、UPDATE、SYNC、WATCHER、GENERIC）。RemoteFilePullTask则是内置的"从远程 URL 拉取文件"基类，基于 context 中的 httpx 客户端实现。

任务注册表位于 backend/tasks/registry.py，SCHEDULED_TASKS与MANUAL_TASKS两个字典把稳定字符串键映射到任务实例。enqueue_task()通过run_task_by_name入队——作业负载里只存任务名而非 pickled 任务对象，这样 Redis 中不依赖任务代码所在的位置，跨版本更稳健。

一个典型范例是 backend/tasks/scheduled/scan_library.py 中的ScanLibraryTask：它继承PeriodicTask，enabled与cron_string直接取自配置（ENABLE_SCHEDULED_RESCAN、SCHEDULED_RESCAN_CRON），超时使用专门的SCAN_TIMEOUT（库扫描不是五分钟能结束的任务），run()中根据各元数据 handler 的is_enabled()状态动态组装启用的元数据源，再调用scan_platforms执行扫描。

数据库迁移（Alembic）

三数据库兼容是硬约束

SKILL 文档强调：迁移必须在 MariaDB、MySQL、PostgreSQL 上都能工作。CI 会在 Postgres 和 MariaDB 上运行alembic upgrade head（迁移工作流见.github/workflows/migrations.yml），且 MySQL 没有 CI 覆盖，更需要谨慎。需要方言差异时使用 batch mode 或数据库特化 SQL；新迁移照抄alembic/versions/中既有迁移的风格。

迁移命令流程

cd backend uv run alembic revision --autogenerate -m "short description" # 生成，然后必须人工审查 uv run alembic upgrade head # 应用 uv run alembic downgrade -1 # 验证降级可用

自动生成的迁移必须人工审查——autogenerate 无法覆盖 server-default、enum、索引细节与跨方言差异。另外，virtual_collections数据库视图被显式排除在迁移之外（它由触发器维护，详见 docs/BACKEND_ARCHITECTURE.md 中关于该视图的说明）。

迁移卫生规范（评审中反复出现的修复项）

SKILL 文档总结了五条来自真实评审反馈的迁移纪律：

1. Rebase 后编号冲突：两个开发分支同时取了下一个编号。rebase 到master后发现你的0102_*已被占用时，重命名文件、更新revision，并把down_revision重新链接到当前真正的前置迁移，然后在新库上跑alembic upgrade head确认链路线性。
2. 优先使用内建幂等标志，而非手工 introspection：op.create_table(..., if_not_exists=True)和create_index(..., if_not_exists=True)优于用inspect(conn).get_table_names()包住整段代码。inspect()只留给内建标志表达不了的场景。
3. 已发布迁移必须能承受部分执行：MySQL/MariaDB 对每条 DDL 语句自动提交，而 Alembic 只在成功后才 stamp，因此一条中途失败的迁移会留下前半段语句，下次启动会从头重放。需要逐步保护，且对op.execute("ALTER TABLE ...")这类裸 SQL 字符串，要用utils.database.column_names过滤。backend/tests/test_migrations.py专门钉住这些重放场景。
4. 未发布的迁移就地修改：只要迁移还没随 tag 发布，就直接修改它，而不是在其上堆叠 fixup 迁移；只有已发布迁移才不可变。
5. 避免可绕过的数据回填：为了归一化解析器现在已能处理的值而重写每一行，是维护负担。优先修解析器，让下一次扫描自然收敛——除非脏行确实用户可见且不可恢复。
6. 能用生成列+索引解决的，不要用 join：当某字段只用于排序或过滤（如generated_first_release_date）时，优先生成列加索引，并在模型的__table_args__中说明，而不是写在注释里。

迁移与模型的漂移守卫

backend/tests/test_migrations.py 中test_no_index_drift_between_models_and_migrations使用 Alembic 的compare_metadata对比迁移后的数据库 schema 与模型元数据，确保每个已迁移索引都声明在模型上、反之亦然——测试库由迁移构建，因此某一边漏了索引，autogenerate 会一直提出删索引，只有此测试能提前暴露。另一个测试则确保POSTGRESQL_FK_INDEXES恰好覆盖所有未索引的外键（MariaDB/MySQL 隐式为外键建索引，PostgreSQL 不会，故需显式补齐）。

OpenAPI → 前端类型管线

FastAPI 在GET /openapi.json提供完整 schema，前端据此重新生成 TypeScript 类型：

# 后端运行在 :3000，然后在 frontend/ 下执行 npm run generate # 通过 openapi-typescript-codegen 写入 src/__generated__/

因此，任何对响应 schema 或路由签名的改动，之后都必须执行npm run generate并做前端 typecheck，否则前端类型与后端接口脱节（前端生成产物位于frontend/src/__generated__/，273 个.ts文件）。

运行、测试与代码质量

开发运行与测试

cd backend uv run python3 main.py # 运行（迁移在启动时自动应用） uv run pytest <path/file> # 只跑受影响的测试文件，绝不要跑整个测试套件

backend/main.py 显示直接运行时依次执行：alembic upgrade head应用迁移 →asyncio.run(main())执行启动任务 →uvicorn.run("main:app", reload=True)启动开发服务器。启动任务（backend/startup.py）会清理过期扫描与旧调度器残留、按需入队回填任务（WebP 转换、save 哈希重算、元数据仓库重建），并把 7 个 JSON fixture 索引（MAME、ScummVM、PS1/PS2/PSP 序列号、BIOS 哈希）加载进 Redis 缓存。

测试体系要点：

• pytest + pytest-asyncio，按pytest-xdistworker 隔离（每个 worker 独立数据库）；
• fakeredis提供内存 Redis；
• pytest-recording的 VCR cassettes 模拟外部 API（见backend/tests/handler/cassettes/）；
• Hypothesis 用于属性测试；
• 测试目录镜像源码布局：backend/tests/<area>/对应backend/<area>/。
• 首次初始化测试数据库：docker exec -i romm-db-dev mariadb -uroot -p<pw> < backend/romm_test/setup.sql（测试库示例数据位于backend/romm_test/，包含 n64、psx、psp 等平台的真实 ROM 文件）。

Lint / 格式 / 类型检查：Trunk

代码质量检查统一通过Trunk编排（ruff、black、isort、mypy、bandit）：

trunk fmt && trunk check

CI 在每个 PR 上强制 Trunk 检查，严禁用--no-verify绕过。

测试纪律

新增或修改的逻辑必须有测试；新增 endpoint 必须有 endpoint 测试。仓库测试覆盖了从 backend/tests/endpoints/（各 API 模块测试）到 backend/tests/handler/（认证、数据库、元数据、文件系统 handler）再到 backend/tests/tasks/（任务注册表、cron 配置、各类清理任务）的完整链路，可作为新测试的编写范本。

总结

RomM 后端的工程规范可以浓缩为三条主线：分层（endpoint 保持薄，业务逻辑在 handler，外部 API 在 typed adapters）、权限（角色 + 细粒度 scope，统一由@protected_route强制）、迁移纪律（三数据库兼容、幂等标志优先、已发布迁移不可变、测试钉住重放行为）。在此基础上，用uv管理环境与依赖、用 Trunk 统一 lint/format/type-check、用npm run generate保持前后端类型同步，即构成了完整的日常开发闭环。任何深入开发之前，建议先通读 docs/BACKEND_ARCHITECTURE.md，再按本指南的分层与约定动手。

【免费下载链接】rommA beautiful, powerful, self-hosted ROM manager and player.项目地址: https://gitcode.com/GitHub_Trending/rom/romm