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

RomM 后端开发实战指南FastAPI / SQLAlchemy 分层架构、认证作用域与 Alembic 迁移规范【免费下载链接】rommA beautiful, powerful, self-hosted ROM manager and player.项目地址: https://gitcode.com/GitHub_Trending/rom/rommRomM 是一个自托管的 ROM 管理与游玩平台其后端是基于 FastAPI 构建的 Python 服务位于仓库backend/目录。本指南以 RomM 官方后端开发技能文档.claude/skills/backend-development/SKILL.md为核心骨架结合仓库源码逐层解析其分层架构、认证与作用域体系、Alembic 迁移卫生规范、OpenAPI→前端类型管线以及 uv/pytest/trunk 工作流。读完本文你将能够在新功能开发时准确判断代码应落在哪个目录、如何保护路由、如何安全地编写数据库迁移并跑通后端测试与代码质量检查。技术栈与运行环境总览RomM 后端技术选型非常集中其核心依赖与职责如下组件选型说明语言Python 3.14全仓库统一使用uv管理依赖与虚拟环境Web 框架FastAPI提供 OpenAPI 文档、Pydantic 校验、异步支持ORMSQLAlchemy 2.0声明式Mapped/mapped_column风格数据库MariaDB默认、MySQL、PostgreSQL三方言需同时兼容迁移Alembic迁移脚本位于backend/alembic/versions/缓存/队列Redis RQ承担会话、缓存、任务队列与房间状态实时通信Socket.IO/ws与/netplay挂载点ASGI 服务器Uvicorn / Gunicorn本地开发与 Docker 部署完整架构参考文档位于 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层中间件职责1CORSMiddleware跨域支持来源由ROMM_CORS_ALLOWED_ORIGINS配置2UploadSizeLimitMiddleware在 multipart 落盘前限制上传体积saves/states/screenshots 与 memory-cards 各有独立上限3CSRFMiddleware基于 cookie header 的 CSRF 防护/api/token、设备配对等 URL 被豁免4AuthenticationMiddlewareHybridAuthBackendBasic、Bearer、Session、OIDC 多方式认证5RedisSessionMiddleware基于 Redis 的 cookie 会话cookie 名为romm_session6set_context_middleware将 aiohttp/httpx 客户端注入 context vars后端编码约定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.pysync_engine create_engine( ConfigManager.get_db_engine(), pool_pre_pingTrue, pool_recycleDB_POOL_RECYCLE_SECONDS, echoFalse, ) sync_session sessionmaker(bindsync_engine, expire_on_commitFalse)开启DEV_SQL_ECHO时还会通过 SQLAlchemy 事件钩子在before_cursor_execute/after_cursor_execute打印每条 SQL 与执行耗时用于开发期排查。BaseModel 的时间戳约定所有 ORM 模型继承 backend/models/base.py 中的BaseModel自动获得created_at与updated_at两个TIMESTAMP(timezoneTrue)列均以 UTC 为准class BaseModel(DeclarativeBase): created_at: Mapped[datetime] mapped_column(TIMESTAMP(timezoneTrue), defaultutc_now) updated_at: Mapped[datetime] mapped_column(TIMESTAMP(timezoneTrue), defaultutc_now, onupdateutc_now)该文件还定义了FILE_NAME_MAX_LENGTH450、FILE_PATH_MAX_LENGTH1000、FILE_EXTENSION_MAX_LENGTH100等常量以及从文件名派生no_tags/no_ext/extension列的辅助函数——这些派生列通过validates钩子与源列保持同步。认证与作用域Auth ScopesRomM 采用角色 细粒度 scope两级权限模型。三级角色角色定义在 backend/models/user.py 中VIEWER只读访问EDITOR在 VIEWER 基础上可写 ROMs / platforms / assetsADMIN在 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())即拥有写权限的角色天然继承所有只读 scopeADMIN 进一步获得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_bearertokenUrl 为/token作为Security依赖注入同时挂载HTTPBasic(auto_errorFalse)支持 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.pyTask抽象类包含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.pySCHEDULED_TASKS与MANUAL_TASKS两个字典把稳定字符串键映射到任务实例。enqueue_task()通过run_task_by_name入队——作业负载里只存任务名而非 pickled 任务对象这样 Redis 中不依赖任务代码所在的位置跨版本更稳健。一个典型范例是 backend/tasks/scheduled/scan_library.py 中的ScanLibraryTask它继承PeriodicTaskenabled与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 文档总结了五条来自真实评审反馈的迁移纪律Rebase 后编号冲突两个开发分支同时取了下一个编号。rebase 到master后发现你的0102_*已被占用时重命名文件、更新revision并把down_revision重新链接到当前真正的前置迁移然后在新库上跑alembic upgrade head确认链路线性。优先使用内建幂等标志而非手工 introspectionop.create_table(..., if_not_existsTrue)和create_index(..., if_not_existsTrue)优于用inspect(conn).get_table_names()包住整段代码。inspect()只留给内建标志表达不了的场景。已发布迁移必须能承受部分执行MySQL/MariaDB 对每条 DDL 语句自动提交而 Alembic 只在成功后才 stamp因此一条中途失败的迁移会留下前半段语句下次启动会从头重放。需要逐步保护且对op.execute(ALTER TABLE ...)这类裸 SQL 字符串要用utils.database.column_names过滤。backend/tests/test_migrations.py专门钉住这些重放场景。未发布的迁移就地修改只要迁移还没随 tag 发布就直接修改它而不是在其上堆叠 fixup 迁移只有已发布迁移才不可变。避免可绕过的数据回填为了归一化解析器现在已能处理的值而重写每一行是维护负担。优先修解析器让下一次扫描自然收敛——除非脏行确实用户可见且不可恢复。能用生成列索引解决的不要用 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, reloadTrue)启动开发服务器。启动任务backend/startup.py会清理过期扫描与旧调度器残留、按需入队回填任务WebP 转换、save 哈希重算、元数据仓库重建并把 7 个 JSON fixture 索引MAME、ScummVM、PS1/PS2/PSP 序列号、BIOS 哈希加载进 Redis 缓存。测试体系要点pytest pytest-asyncio按pytest-xdistworker 隔离每个 worker 独立数据库fakeredis提供内存 Redispytest-recording的 VCR cassettes 模拟外部 API见backend/tests/handler/cassettes/Hypothesis 用于属性测试测试目录镜像源码布局backend/tests/area/对应backend/area/。首次初始化测试数据库docker exec -i romm-db-dev mariadb -uroot -ppw backend/romm_test/setup.sql测试库示例数据位于backend/romm_test/包含 n64、psx、psp 等平台的真实 ROM 文件。Lint / 格式 / 类型检查Trunk代码质量检查统一通过Trunk编排ruff、black、isort、mypy、bandittrunk fmt trunk checkCI 在每个 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考