维护一场欧罗巴联赛的比赛数据听起来只是把主队、客队、比分放进表格里。真正做起来会发现问题都出在细节上同一支球队在不同文件里叫“安德莱赫特”还是“AND”比赛时间按哪个时区记录比分被覆盖后如何还原第三方数据重复导入时如何避免产生两场相同的比赛。这些问题在没有数据结构约束和接口约束时会随着比赛数量增加而迅速放大。这篇文章从一个具体场景出发在赛事数据系统里维护“安德莱赫特 vs 塞萨洛尼基”这场欧罗巴比赛包括创建球队、创建赛程、记录比分事件、查询比赛详情。技术实现使用 FastAPI SQLAlchemy SQLite涉及表结构设计、状态机设计、幂等处理、时区处理和常见问题排查。整篇文章既可以作为体育数据类应用的入门项目也可以作为后端工程师整理赛事领域建模的参考。1. 赛事数据管理系统解决什么问题从一份欧罗巴赛程说起1.1 为什么手工表格撑不住赛事数据维护很多体育数据项目一开始都是从一个 Excel 文件开始的。维护者按比赛日期新建一行手动填入主队、客队、比分和比赛状态。比赛少的时候没有问题一旦联赛进入资格赛、小组赛、淘汰赛并行阶段问题就开始暴露。常见的手工维护问题包括队名不统一。同一个球队有人写“安德莱赫特”有人写“Anderlecht”也有人写简称“AND”。同一场比赛重复存在。第三方数据源推送一次运营人员又手动录入一次系统里出现两条主客队相同的记录。比分只能记住最终值。比赛过程中出现 1:0、1:1、2:1 的变化最终表里只有一个 2:1过几天没有人能说清楚第二个进球发生在第几分钟。比赛状态靠人工标记。开球后忘了改为“进行中”比赛结束很久还停在“未开始”影响下游统计和展示。这些问题本质上是缺少数据建模和约束。赛事数据管理系统要做的就是让球队、比赛、比分、状态都有明确的结构和规则让数据在录入阶段就被校验而不是等查询阶段才发现错误。1.2 核心模块球队、赛程、比分、查询一个最小可用的赛事数据系统可以拆成四个模块。球队主数据负责维护参赛队伍重点是名称唯一性和基础信息。赛程管理负责创建比赛记录比赛双方、所属赛事、轮次、比赛时间。比分记录负责跟踪比赛过程中的每一次比分变化。查询服务负责把比赛详情、当前比分、球队信息组装后返回给调用方。这四个模块并不复杂但它们是后续做积分榜、射手榜、赛程日历、数据统计的基础。如果这一层的数据口径有问题上层的所有统计都会失真。1.3 技术方案和项目结构示例采用 FastAPI SQLAlchemy SQLite。选择这组技术的原因有三点FastAPI 学习成本低能快速提供 REST 接口适合作为内部运营后台或数据管理服务。SQLAlchemy 同时支持 SQLite、PostgreSQL、MySQL学习环境用 SQLite 零部署生产环境可以平滑切换。Pydantic 负责请求参数校验可以在进入业务逻辑之前拦截非法数据。项目结构如下sports_data/ ├── main.py # FastAPI 入口和路由 ├── models.py # SQLAlchemy ORM 模型 ├── schemas.py # Pydantic 请求响应模型 └── database.py # 数据库连接和会话管理这是一个最小的单服务结构。后续引入迁移工具、定时任务、缓存层时可以按模块继续拆分。2. 数据模型设计先定状态机再建表2.1 比赛状态机让数据流转有规则比赛数据里最容易乱的是状态。如果状态可以随便填下游在计算“未开始比赛数量”时就会混入已经结束的场次。因此要先定义状态机再用数据库约束保证状态值合法。状态含义进入条件后续状态scheduled未开始创建比赛时默认in_progress / postponed / cancelledin_progress进行中开球后通过接口更新finished / postponed / cancelledfinished已结束常规时间或官方判定结束终态需要修正时走修正接口postponed延期赛前或赛中出现延期scheduled / cancelledcancelled取消比赛取消终态比分清空这里有一个容易被忽略的点finished是终态但不代表数据永远不可以修正。足球比赛中可能出现官方更正进球归属、补时时间调整等场景。正确做法是保留原始比分事件允许管理员通过专门的修正接口调整最终结果而不是直接修改事件记录。注意状态字段不要使用无约束的字符串。至少要在数据库层面加 CHECK 约束在应用层再用枚举或常量类管理否则很快就会出现拼写错误导致的脏状态。2.2 球队表和比赛表核心字段如何设计球队表的核心是名称唯一性。同一个球队可以有中文名、英文名、简称但系统内部必须有一个唯一键用来关联比赛。推荐使用英文全称或官方标识作为唯一名称把中文名和简称作为辅助字段。比赛表则要承载赛事上下文。competition表示赛事名称round_name表示轮次external_id用来关联第三方数据源。external_id必须加唯一约束这是避免重复导入的关键。比赛时间字段建议设计为可排序的DATETIME类型不要使用字符串。否则后续按日期筛选比赛时SQL 比较会非常难写。2.3 比分事件表为什么不能只存几个数字很多初版设计会把home_score和away_score直接放在比赛表里每次进球就 update 一次。这个设计的最大问题是丢失历史。假设系统里只存最终比分 2:1当运营人员需要回答“主队的第二个进球是什么时候进的”时数据是缺失的。如果用事件表记录每一次比分变化就能得到完整的时间线第 23 分钟主队进球比分变为 1:0。第 41 分钟客队进球比分变为 1:1。第 67 分钟主队进球比分变为 2:1。比分事件表还有另一个作用防止并发更新把比分覆盖错。直接更新比赛表时如果两个请求同时写入后写入的请求会覆盖前一个结果。将每次变更作为一条独立记录插入再更新比赛表的当前比分可以保证事件可追溯当前比分只是“最新一条事件”的冗余展示。2.4 表关系与 DDL 设计三张表的关系如下teams与matches是一对多关系一场比赛有主队和客队两个外键。matches与match_score_events是一对多关系一场比赛有多条比分事件。对应的 SQLite DDL 如下CREATE TABLE teams ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR(100) NOT NULL UNIQUE, short_name VARCHAR(50), country VARCHAR(50), created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE matches ( id INTEGER PRIMARY KEY AUTOINCREMENT, external_id VARCHAR(100) UNIQUE, competition VARCHAR(100) NOT NULL, round_name VARCHAR(100), home_team_id INTEGER NOT NULL, away_team_id INTEGER NOT NULL, match_time DATETIME NOT NULL, status VARCHAR(20) NOT NULL DEFAULT scheduled, current_home_score INTEGER NOT NULL DEFAULT 0, current_away_score INTEGER NOT NULL DEFAULT 0, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (home_team_id) REFERENCES teams(id), FOREIGN KEY (away_team_id) REFERENCES teams(id), CHECK (home_team_id away_team_id), CHECK (status IN (scheduled, in_progress, finished, postponed, cancelled)), CHECK (current_home_score 0), CHECK (current_away_score 0) ); CREATE TABLE match_score_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, match_id INTEGER NOT NULL, event_minute INTEGER, home_score INTEGER NOT NULL, away_score INTEGER NOT NULL, event_type VARCHAR(20) NOT NULL DEFAULT score, note VARCHAR(255), created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (match_id) REFERENCES matches(id) );这个 DDL 里值得注意的设计点有三个。第一external_id建了唯一索引但允许为空。学习环境下手工创建的比赛没有外部 ID多条空值不会触发唯一约束冲突这是 SQLite 和多数数据库对 NULL 的处理规则。第二matches表增加了CHECK (home_team_id away_team_id)。虽然代码层面会校验主客队不同但数据库约束是最后一道防线能防止脏数据被直接写入。第三match_score_events表只记录进球后的比分不直接修改事件本身。当比分从 1:0 变成 1:1 时新增一行记录即可。3. 用 FastAPI 实现最小可用的赛事数据服务3.1 初始化项目与依赖示例代码基于 Python 3.10因为会用到str | None这种类型注解语法。需要安装的依赖如下pip install fastapi uvicorn sqlalchemy启动和调试时使用 uvicornuvicorn main:app --reload如果原始环境中的 Python 版本低于 3.10需要把模型和 Pydantic 里的str | None改成Optional[str]否则会直接报语法错误。3.2 数据库连接与会话管理database.py负责创建引擎和会话工厂from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase DATABASE_URL sqlite:///./sports.db engine create_engine( DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) class Base(DeclarativeBase): passSQLite 默认不允许跨线程访问同一个连接。check_same_threadFalse是 FastAPI 多线程访问 SQLite 时常用的配置生产环境切换到 PostgreSQL 后可以删除这个参数。3.3 ORM 模型定义models.py中定义三张表对应的 ORM 模型from datetime import datetime from sqlalchemy import ( String, Integer, DateTime, ForeignKey, CheckConstraint ) from sqlalchemy.orm import Mapped, mapped_column, relationship from database import Base class Team(Base): __tablename__ teams id: Mapped[int] mapped_column(primary_keyTrue, indexTrue) name: Mapped[str] mapped_column(String(100), uniqueTrue, indexTrue) short_name: Mapped[str | None] mapped_column(String(50), nullableTrue) country: Mapped[str | None] mapped_column(String(50), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) home_matches: Mapped[list[FootballMatch]] relationship( foreign_keysFootballMatch.home_team_id, back_populateshome_team ) away_matches: Mapped[list[FootballMatch]] relationship( foreign_keysFootballMatch.away_team_id, back_populatesaway_team ) class FootballMatch(Base): __tablename__ matches __table_args__ ( CheckConstraint(home_team_id away_team_id, nameck_match_teams_diff), CheckConstraint( status IN (scheduled, in_progress, finished, postponed, cancelled), nameck_match_status ), ) id: Mapped[int] mapped_column(primary_keyTrue, indexTrue) external_id: Mapped[str | None] mapped_column(String(100), uniqueTrue, nullableTrue, indexTrue) competition: Mapped[str] mapped_column(String(100), indexTrue) round_name: Mapped[str | None] mapped_column(String(100), nullableTrue) home_team_id: Mapped[int] mapped_column(ForeignKey(teams.id)) away_team_id: Mapped[int] mapped_column(ForeignKey(teams.id)) match_time: Mapped[datetime] mapped_column(DateTime, indexTrue) status: Mapped[str] mapped_column(String(20), defaultscheduled, indexTrue) current_home_score: Mapped[int] mapped_column(Integer, default0) current_away_score: Mapped[int] mapped_column(Integer, default0) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) updated_at: Mapped[datetime] mapped_column( DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow ) home_team: Mapped[Team] relationship( foreign_keys[home_team_id], back_populateshome_matches ) away_team: Mapped[Team] relationship( foreign_keys[away_team_id], back_populatesaway_matches ) score_events: Mapped[list[MatchScoreEvent]] relationship( back_populatesmatch, cascadeall, delete-orphan ) class MatchScoreEvent(Base): __tablename__ match_score_events id: Mapped[int] mapped_column(primary_keyTrue, indexTrue) match_id: Mapped[int] mapped_column(ForeignKey(matches.id), indexTrue) event_minute: Mapped[int | None] mapped_column(Integer, nullableTrue) home_score: Mapped[int] mapped_column(Integer) away_score: Mapped[int] mapped_column(Integer) event_type: Mapped[str] mapped_column(String(20), defaultscore) note: Mapped[str | None] mapped_column(String(255), nullableTrue) created_at: Mapped[datetime] mapped_column(DateTime, defaultdatetime.utcnow) match: Mapped[FootballMatch] relationship(back_populatesscore_events)这段代码的关键点有三个。第一FootballMatch类名没有使用Match是为了避免与 Python 的match关键字在阅读时产生歧义。第二score_events关系配置了cascadeall, delete-orphan删除比赛时会同时删除它的比分事件避免产生孤儿数据。第三updated_at使用onupdatedatetime.utcnow每次更新比赛记录时自动刷新更新时间。这个字段在排查“比分为什么被修改”时非常有用。3.4 Pydantic 请求响应模型schemas.py定义接口的入参和出参from datetime import datetime from pydantic import BaseModel, ConfigDict, Field class TeamCreate(BaseModel): name: str Field(min_length1, max_length100) short_name: str | None Field(defaultNone, max_length50) country: str | None Field(defaultNone, max_length50) class TeamOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int name: str short_name: str | None country: str | None created_at: datetime class MatchCreate(BaseModel): external_id: str | None Field(defaultNone, max_length100) competition: str Field(min_length1, max_length100) round_name: str | None Field(defaultNone, max_length100) home_team_id: int away_team_id: int match_time: datetime class ScoreEventCreate(BaseModel): event_minute: int | None Field(defaultNone, ge0, le130) home_score: int Field(ge0) away_score: int Field(ge0) event_type: str Field(defaultscore, max_length20) note: str | None Field(defaultNone, max_length255) class MatchOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int external_id: str | None competition: str round_name: str | None home_team: TeamOut away_team: TeamOut match_time: datetime status: str current_home_score: int current_away_score: int created_at: datetime updated_at: datetimeField(ge0)会在请求进入业务逻辑之前校验比分不能为负数。event_minute限制在 0 到 130 之间能拦截明显不合理的分钟数。3.5 FastAPI 路由与业务逻辑main.py中实现创建球队、创建比赛、记录比分、结束比赛、查询比赛详情五个接口from datetime import datetime, timezone from fastapi import FastAPI, Depends, HTTPException from sqlalchemy import select from sqlalchemy.orm import Session, selectinload from database import SessionLocal, engine, Base from models import Team, FootballMatch, MatchScoreEvent from schemas import ( TeamCreate, TeamOut, MatchCreate, MatchOut, ScoreEventCreate ) Base.metadata.create_all(bindengine) app FastAPI(titleFootball Match Data API) def get_db(): db SessionLocal() try: yield db finally: db.close() def to_utc_naive(dt: datetime) - datetime: if dt.tzinfo is None: return dt return dt.astimezone(timezone.utc).replace(tzinfoNone) def get_match_or_404(db: Session, match_id: int) - FootballMatch: stmt ( select(FootballMatch) .options( selectinload(FootballMatch.home_team), selectinload(FootballMatch.away_team), ) .where(FootballMatch.id match_id) ) match db.execute(stmt).scalar_one_or_none() if match is None: raise HTTPException(status_code404, detailmatch not found) return match app.post(/teams, response_modelTeamOut) def create_team(payload: TeamCreate, db: Session Depends(get_db)): exists db.execute( select(Team).where(Team.name payload.name) ).scalar_one_or_none() if exists: raise HTTPException(status_code400, detailteam name already exists) team Team(**payload.model_dump()) db.add(team) db.commit() db.refresh(team) return team app.post(/matches, response_modelMatchOut) def create_match(payload: MatchCreate, db: Session Depends(get_db)): home db.get(Team, payload.home_team_id) away db.get(Team, payload.away_team_id) if home is None or away is None: raise HTTPException(