維護(hù)一場(chǎng)歐羅巴聯(lián)賽的比賽數(shù)據(jù)聽(tīng)起來(lái)只是把主隊(duì)、客隊(duì)、比分放進(jìn)表格里。真正做起來(lái)會(huì)發(fā)現(xiàn)問(wèn)題都出在細(xì)節(jié)上同一支球隊(duì)在不同文件里叫“安德萊赫特”還是“AND”比賽時(shí)間按哪個(gè)時(shí)區(qū)記錄比分被覆蓋后如何還原第三方數(shù)據(jù)重復(fù)導(dǎo)入時(shí)如何避免產(chǎn)生兩場(chǎng)相同的比賽。這些問(wèn)題在沒(méi)有數(shù)據(jù)結(jié)構(gòu)約束和接口約束時(shí)會(huì)隨著比賽數(shù)量增加而迅速放大。這篇文章從一個(gè)具體場(chǎng)景出發(fā)在賽事數(shù)據(jù)系統(tǒng)里維護(hù)“安德萊赫特 vs 塞薩洛尼基”這場(chǎng)歐羅巴比賽包括創(chuàng)建球隊(duì)、創(chuàng)建賽程、記錄比分事件、查詢比賽詳情。技術(shù)實(shí)現(xiàn)使用 FastAPI SQLAlchemy SQLite涉及表結(jié)構(gòu)設(shè)計(jì)、狀態(tài)機(jī)設(shè)計(jì)、冪等處理、時(shí)區(qū)處理和常見(jiàn)問(wèn)題排查。整篇文章既可以作為體育數(shù)據(jù)類應(yīng)用的入門項(xiàng)目也可以作為后端工程師整理賽事領(lǐng)域建模的參考。1. 賽事數(shù)據(jù)管理系統(tǒng)解決什么問(wèn)題從一份歐羅巴賽程說(shuō)起1.1 為什么手工表格撐不住賽事數(shù)據(jù)維護(hù)很多體育數(shù)據(jù)項(xiàng)目一開(kāi)始都是從一個(gè) Excel 文件開(kāi)始的。維護(hù)者按比賽日期新建一行手動(dòng)填入主隊(duì)、客隊(duì)、比分和比賽狀態(tài)。比賽少的時(shí)候沒(méi)有問(wèn)題一旦聯(lián)賽進(jìn)入資格賽、小組賽、淘汰賽并行階段問(wèn)題就開(kāi)始暴露。常見(jiàn)的手工維護(hù)問(wèn)題包括隊(duì)名不統(tǒng)一。同一個(gè)球隊(duì)有人寫(xiě)“安德萊赫特”有人寫(xiě)“Anderlecht”也有人寫(xiě)簡(jiǎn)稱“AND”。同一場(chǎng)比賽重復(fù)存在。第三方數(shù)據(jù)源推送一次運(yùn)營(yíng)人員又手動(dòng)錄入一次系統(tǒng)里出現(xiàn)兩條主客隊(duì)相同的記錄。比分只能記住最終值。比賽過(guò)程中出現(xiàn) 1:0、1:1、2:1 的變化最終表里只有一個(gè) 2:1過(guò)幾天沒(méi)有人能說(shuō)清楚第二個(gè)進(jìn)球發(fā)生在第幾分鐘。比賽狀態(tài)靠人工標(biāo)記。開(kāi)球后忘了改為“進(jìn)行中”比賽結(jié)束很久還停在“未開(kāi)始”影響下游統(tǒng)計(jì)和展示。這些問(wèn)題本質(zhì)上是缺少數(shù)據(jù)建模和約束。賽事數(shù)據(jù)管理系統(tǒng)要做的就是讓球隊(duì)、比賽、比分、狀態(tài)都有明確的結(jié)構(gòu)和規(guī)則讓數(shù)據(jù)在錄入階段就被校驗(yàn)而不是等查詢階段才發(fā)現(xiàn)錯(cuò)誤。1.2 核心模塊球隊(duì)、賽程、比分、查詢一個(gè)最小可用的賽事數(shù)據(jù)系統(tǒng)可以拆成四個(gè)模塊。球隊(duì)主數(shù)據(jù)負(fù)責(zé)維護(hù)參賽隊(duì)伍重點(diǎn)是名稱唯一性和基礎(chǔ)信息。賽程管理負(fù)責(zé)創(chuàng)建比賽記錄比賽雙方、所屬賽事、輪次、比賽時(shí)間。比分記錄負(fù)責(zé)跟蹤比賽過(guò)程中的每一次比分變化。查詢服務(wù)負(fù)責(zé)把比賽詳情、當(dāng)前比分、球隊(duì)信息組裝后返回給調(diào)用方。這四個(gè)模塊并不復(fù)雜但它們是后續(xù)做積分榜、射手榜、賽程日歷、數(shù)據(jù)統(tǒng)計(jì)的基礎(chǔ)。如果這一層的數(shù)據(jù)口徑有問(wèn)題上層的所有統(tǒng)計(jì)都會(huì)失真。1.3 技術(shù)方案和項(xiàng)目結(jié)構(gòu)示例采用 FastAPI SQLAlchemy SQLite。選擇這組技術(shù)的原因有三點(diǎn)FastAPI 學(xué)習(xí)成本低能快速提供 REST 接口適合作為內(nèi)部運(yùn)營(yíng)后臺(tái)或數(shù)據(jù)管理服務(wù)。SQLAlchemy 同時(shí)支持 SQLite、PostgreSQL、MySQL學(xué)習(xí)環(huán)境用 SQLite 零部署生產(chǎn)環(huán)境可以平滑切換。Pydantic 負(fù)責(zé)請(qǐng)求參數(shù)校驗(yàn)可以在進(jìn)入業(yè)務(wù)邏輯之前攔截非法數(shù)據(jù)。項(xiàng)目結(jié)構(gòu)如下sports_data/ ├── main.py # FastAPI 入口和路由 ├── models.py # SQLAlchemy ORM 模型 ├── schemas.py # Pydantic 請(qǐng)求響應(yīng)模型 └── database.py # 數(shù)據(jù)庫(kù)連接和會(huì)話管理這是一個(gè)最小的單服務(wù)結(jié)構(gòu)。后續(xù)引入遷移工具、定時(shí)任務(wù)、緩存層時(shí)可以按模塊繼續(xù)拆分。2. 數(shù)據(jù)模型設(shè)計(jì)先定狀態(tài)機(jī)再建表2.1 比賽狀態(tài)機(jī)讓數(shù)據(jù)流轉(zhuǎn)有規(guī)則比賽數(shù)據(jù)里最容易亂的是狀態(tài)。如果狀態(tài)可以隨便填下游在計(jì)算“未開(kāi)始比賽數(shù)量”時(shí)就會(huì)混入已經(jīng)結(jié)束的場(chǎng)次。因此要先定義狀態(tài)機(jī)再用數(shù)據(jù)庫(kù)約束保證狀態(tài)值合法。狀態(tài)含義進(jìn)入條件后續(xù)狀態(tài)scheduled未開(kāi)始創(chuàng)建比賽時(shí)默認(rèn)in_progress / postponed / cancelledin_progress進(jìn)行中開(kāi)球后通過(guò)接口更新finished / postponed / cancelledfinished已結(jié)束常規(guī)時(shí)間或官方判定結(jié)束終態(tài)需要修正時(shí)走修正接口postponed延期賽前或賽中出現(xiàn)延期scheduled / cancelledcancelled取消比賽取消終態(tài)比分清空這里有一個(gè)容易被忽略的點(diǎn)finished是終態(tài)但不代表數(shù)據(jù)永遠(yuǎn)不可以修正。足球比賽中可能出現(xiàn)官方更正進(jìn)球歸屬、補(bǔ)時(shí)時(shí)間調(diào)整等場(chǎng)景。正確做法是保留原始比分事件允許管理員通過(guò)專門的修正接口調(diào)整最終結(jié)果而不是直接修改事件記錄。注意狀態(tài)字段不要使用無(wú)約束的字符串。至少要在數(shù)據(jù)庫(kù)層面加 CHECK 約束在應(yīng)用層再用枚舉或常量類管理否則很快就會(huì)出現(xiàn)拼寫(xiě)錯(cuò)誤導(dǎo)致的臟狀態(tài)。2.2 球隊(duì)表和比賽表核心字段如何設(shè)計(jì)球隊(duì)表的核心是名稱唯一性。同一個(gè)球隊(duì)可以有中文名、英文名、簡(jiǎn)稱但系統(tǒng)內(nèi)部必須有一個(gè)唯一鍵用來(lái)關(guān)聯(lián)比賽。推薦使用英文全稱或官方標(biāo)識(shí)作為唯一名稱把中文名和簡(jiǎn)稱作為輔助字段。比賽表則要承載賽事上下文。competition表示賽事名稱round_name表示輪次external_id用來(lái)關(guān)聯(lián)第三方數(shù)據(jù)源。external_id必須加唯一約束這是避免重復(fù)導(dǎo)入的關(guān)鍵。比賽時(shí)間字段建議設(shè)計(jì)為可排序的DATETIME類型不要使用字符串。否則后續(xù)按日期篩選比賽時(shí)SQL 比較會(huì)非常難寫(xiě)。2.3 比分事件表為什么不能只存幾個(gè)數(shù)字很多初版設(shè)計(jì)會(huì)把home_score和away_score直接放在比賽表里每次進(jìn)球就 update 一次。這個(gè)設(shè)計(jì)的最大問(wèn)題是丟失歷史。假設(shè)系統(tǒng)里只存最終比分 2:1當(dāng)運(yùn)營(yíng)人員需要回答“主隊(duì)的第二個(gè)進(jìn)球是什么時(shí)候進(jìn)的”時(shí)數(shù)據(jù)是缺失的。如果用事件表記錄每一次比分變化就能得到完整的時(shí)間線第 23 分鐘主隊(duì)進(jìn)球比分變?yōu)?1:0。第 41 分鐘客隊(duì)進(jìn)球比分變?yōu)?1:1。第 67 分鐘主隊(duì)進(jìn)球比分變?yōu)?2:1。比分事件表還有另一個(gè)作用防止并發(fā)更新把比分覆蓋錯(cuò)。直接更新比賽表時(shí)如果兩個(gè)請(qǐng)求同時(shí)寫(xiě)入后寫(xiě)入的請(qǐng)求會(huì)覆蓋前一個(gè)結(jié)果。將每次變更作為一條獨(dú)立記錄插入再更新比賽表的當(dāng)前比分可以保證事件可追溯當(dāng)前比分只是“最新一條事件”的冗余展示。2.4 表關(guān)系與 DDL 設(shè)計(jì)三張表的關(guān)系如下teams與matches是一對(duì)多關(guān)系一場(chǎng)比賽有主隊(duì)和客隊(duì)兩個(gè)外鍵。matches與match_score_events是一對(duì)多關(guān)系一場(chǎng)比賽有多條比分事件。對(duì)應(yīng)的 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) );這個(gè) DDL 里值得注意的設(shè)計(jì)點(diǎn)有三個(gè)。第一external_id建了唯一索引但允許為空。學(xué)習(xí)環(huán)境下手工創(chuàng)建的比賽沒(méi)有外部 ID多條空值不會(huì)觸發(fā)唯一約束沖突這是 SQLite 和多數(shù)數(shù)據(jù)庫(kù)對(duì) NULL 的處理規(guī)則。第二matches表增加了CHECK (home_team_id away_team_id)。雖然代碼層面會(huì)校驗(yàn)主客隊(duì)不同但數(shù)據(jù)庫(kù)約束是最后一道防線能防止臟數(shù)據(jù)被直接寫(xiě)入。第三match_score_events表只記錄進(jìn)球后的比分不直接修改事件本身。當(dāng)比分從 1:0 變成 1:1 時(shí)新增一行記錄即可。3. 用 FastAPI 實(shí)現(xiàn)最小可用的賽事數(shù)據(jù)服務(wù)3.1 初始化項(xiàng)目與依賴示例代碼基于 Python 3.10因?yàn)闀?huì)用到str | None這種類型注解語(yǔ)法。需要安裝的依賴如下pip install fastapi uvicorn sqlalchemy啟動(dòng)和調(diào)試時(shí)使用 uvicornuvicorn main:app --reload如果原始環(huán)境中的 Python 版本低于 3.10需要把模型和 Pydantic 里的str | None改成Optional[str]否則會(huì)直接報(bào)語(yǔ)法錯(cuò)誤。3.2 數(shù)據(jù)庫(kù)連接與會(huì)話管理database.py負(fù)責(zé)創(chuàng)建引擎和會(huì)話工廠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 默認(rèn)不允許跨線程訪問(wèn)同一個(gè)連接。check_same_threadFalse是 FastAPI 多線程訪問(wèn) SQLite 時(shí)常用的配置生產(chǎn)環(huán)境切換到 PostgreSQL 后可以刪除這個(gè)參數(shù)。3.3 ORM 模型定義models.py中定義三張表對(duì)應(yīng)的 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)這段代碼的關(guān)鍵點(diǎn)有三個(gè)。第一FootballMatch類名沒(méi)有使用Match是為了避免與 Python 的match關(guān)鍵字在閱讀時(shí)產(chǎn)生歧義。第二score_events關(guān)系配置了cascadeall, delete-orphan刪除比賽時(shí)會(huì)同時(shí)刪除它的比分事件避免產(chǎn)生孤兒數(shù)據(jù)。第三updated_at使用onupdatedatetime.utcnow每次更新比賽記錄時(shí)自動(dòng)刷新更新時(shí)間。這個(gè)字段在排查“比分為什么被修改”時(shí)非常有用。3.4 Pydantic 請(qǐng)求響應(yīng)模型schemas.py定義接口的入?yún)⒑统鰠rom 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)會(huì)在請(qǐng)求進(jìn)入業(yè)務(wù)邏輯之前校驗(yàn)比分不能為負(fù)數(shù)。event_minute限制在 0 到 130 之間能攔截明顯不合理的分鐘數(shù)。3.5 FastAPI 路由與業(yè)務(wù)邏輯main.py中實(shí)現(xiàn)創(chuàng)建球隊(duì)、創(chuàng)建比賽、記錄比分、結(jié)束比賽、查詢比賽詳情五個(gè)接口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(