3天手写实现公交车app，告别看教程不会写的尴尬
是不是也这样？B站收藏了99+个Python项目，CSDN存了上百篇架构设计，结果真要动手写个公交查询系统，脑子一片空白。卡在“需求拆解”这一步，连数据库表都建不起来。
今天不灌鸡汤，直接带你从零手写实现一个最小可用的公交车app后端。不是那种大而全的演示代码，而是能跑通、能扩展、面试能讲清楚的实战逻辑。跟着敲一遍，你会发现“写项目”没那么玄乎。
项目目标与边界界定
很多新人一上来就想做“高德地图”，结果第一周就崩了。我们得先砍掉90%的需求。
这个手写实现的公交车app，核心功能只有三个：
查询某条线路的所有站点及顺序。
查询某两个站点之间的换乘方案。
获取实时到站预测（模拟数据）。
不做什么同样重要：不做GPS实时定位、不做用户登录鉴权、不做支付、不做地图渲染。这些是前端和第三方API的事，后端只负责数据和逻辑。
为什么这么定？因为初学者最大的坑就是“过度设计”。你在CSDN上搜“公交系统设计”，看到的往往是百万行代码的工业级方案，对你毫无参考价值。我们要的是“最小闭环”，先把数据流转跑通。
目录结构：像搭积木一样组织代码
别把代码全塞在main.py里。哪怕只有500行，也要分层。这是工程化的第一步，也是面试官看重的“代码洁癖”。
bus_app/
├── main.py # 入口，启动FastAPI服务
├── models/ # 数据模型，对应数据库表
│ ├── __init__.py
│ └── bus.py # 线路、站点、班次定义
├── services/ # 业务逻辑层，核心算法在这里
│ ├── __init__.py
│ └── route.py # 换乘路径计算逻辑
├── database/ # 数据库连接与初始化
│ ├── __init__.py
│ ├── db.py # SQLAlchemy连接池
│ └── seed.py # 初始测试数据插入
└── requirements.txt # 依赖管理
关键点：services层不直接操作数据库，它调用models或database提供的方法。这样以后你想把MySQL换成PostgreSQL，只改database层，业务逻辑一行不动。这就是解耦，面试时能讲出这一层，比背八股文有用得多。
核心代码实现：逐行拆解
1. 数据模型：别乱写字段名
很多人用dict存数据，看似灵活，实则致命。结构化数据必须用Pydantic或SQLAlchemy ORM。
# models/bus.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from database.db import Base
class Station(Base):
__tablename__ = 'stations'
id = Column(Integer, primary_key=True)
name = Column(String(50), unique=True, nullable=False)
latitude = Column(Float) # 经纬度用于后续距离计算
longitude = Column(Float)
class Route(Base):
__tablename__ = 'routes'
id = Column(Integer, primary_key=True)
name = Column(String(20), unique=True, nullable=False) # 如 K1路
description = Column(String(200))
class RouteStop(Base):
多对多关系：一条线路有多个站点，一个站点属于多条线路
__tablename__ = 'route_stops'
id = Column(Integer, primary_key=True)
route_id = Column(Integer, ForeignKey('routes.id'))
station_id = Column(Integer, ForeignKey('stations.id'))
stop_order = Column(Integer, nullable=False) # 关键：站点顺序！
route = relationship(Route)
station = relationship(Station)
注意stop_order字段。很多新手会漏掉这个，导致无法判断“下一站”。公交是有序的，不是无序集合。
2. 换乘算法：别用暴力递归
求两点间最短换乘，本质是图的最短路径问题。节点是站点，边是“同一条线路内相邻站点”或“同站换乘”。
# services/route.py
from collections import deque
from models.bus import RouteStop, Station, Route
from database.db import Session
def find_min_transfers(start_name: str, end_name: str, db: Session) - list:
使用BFS寻找最少换乘次数路径
返回: [站点1, 线路1, 站点2, 线路2, ..., 终点]
# 1. 构建邻接表：站点 - [(下一站, 线路名), (换乘线路1), (换乘线路2)...]
graph = {}
stops = db.query(RouteStop).all()
# 初始化所有站点
stations = db.query(Station).all()
for s in stations:
graph[s.name] = []
# 构建同线路相邻关系 + 同站换乘关系
for stop in stops:
# 同线路相邻：当前站 - 下一站
next_stop = db.query(RouteStop).filter_by(
route_id=stop.route_id,
stop_order=stop.stop_order + 1
).first()
if next_stop:
graph[stop.station.name].append((next_stop.station.name, stop.route.name))
# 同站换乘：当前站的其他线路
other_routes = db.query(RouteStop).filter_by(
station_id=stop.station_id
).filter(RouteStop.route_id != stop.route_id).all()
for other in other_routes:
graph[stop.station.name].append((stop.station.name, other.route.name))
# 2. BFS搜索
queue = deque()
queue.append((start_name, [], start_name)) # (当前站, 已走路径, 起始站)
visited = {start_name}
while queue:
current, path, origin = queue.popleft()
if current == end_name:
return path + [current]
for next_station, route_name in graph[current]:
if next_station not in visited:
visited.add(next_station)
# 记录路径：站点 + 经过的线路
new_path = path + [route_name, next_station]
queue.append((next_station, new_path, origin))
return [] # 无路径
逐行解读：
graph 不是简单的站点连站点，而是包含了“通过哪条线路到达”。
BFS保证第一次找到终点时，就是换乘次数最少的路径（假设每站时间成本相同）。
visited 防止死循环，这是图论算法的基本功。
3. 实时到站预测：模拟而非真实
真实公交到站需要GPS数据流，这里我们用“固定间隔+随机误差”模拟。
# services/route.py 补充
import random
def predict_arrival(station_name: str, db: Session) - dict:
模拟到站预测：基于历史平均间隔
假设每条线路每10分钟一班，误差±2分钟
# 实际项目中，这里应查询最近一辆车的GPS位置
# 模拟：随机生成一个1-15分钟的等待时间
wait_time = random.randint(1, 15)
routes = db.query(Route).filter(Route.id.in_(
db.query(RouteStop.route_id).filter_by(
station_id=db.query(Station.id).filter_by(name=station_name).first()
)
)).all()
predictions = []
for r in routes:
predictions.append({
route: r.name,
eta_minutes: wait_time + random.randint(-2, 2),
status: running
})
return predictions
避坑：不要在这里写复杂的机器学习模型预测到站。那是算法工程师的事。后端先保证接口稳定返回结构化的JSON，前端再决定怎么展示。
运行与测试：别等上线才修Bug
1. 初始化测试数据
# database/seed.py
from database.db import SessionLocal
from models.bus import Station, Route, RouteStop
def seed_data():
db = SessionLocal()
# 清空旧数据
db.query(RouteStop).delete()
db.query(Route).delete()
db.query(Station).delete()
# 插入站点
s1 = Station(name=人民广场, latitude=31.23, longitude=121.47)
s2 = Station(name=南京路, latitude=31.24, longitude=121.48)
s3 = Station(name=陆家嘴, latitude=31.25, longitude=121.49)
db.add_all([s1, s2, s3])
db.commit()
# 插入线路
r1 = Route(name=K1路, description=市中心环线)
r2 = Route(name=K2路, description=东西向干线)
db.add_all([r1, r2])
db.commit()
# 插入站点顺序
# K1: 人民广场 - 南京路 - 陆家嘴
# K2: 人民广场 - 陆家嘴 (假设K2也经过，形成换乘)
db.add(RouteStop(route_id=r1.id, station_id=s1.id, stop_order=1))
db.add(RouteStop(route_id=r1.id, station_id=s2.id, stop_order=2))
db.add(RouteStop(route_id=r1.id, station_id=s3.id, stop_order=3))
db.add(RouteStop(route_id=r2.id, station_id=s1.id, stop_order=1))
db.add(RouteStop(route_id=r2.id, station_id=s3.id, stop_order=2))
db.commit()
db.close()
2. 写测试用例
用pytest，不要只靠print调试。
# tests/test_route.py
import pytest
from database.db import SessionLocal
from services.route import find_min_transfers
def test_min_transfers():
db = SessionLocal()
# 从人民广场到陆家嘴，K1直达，应返回K1
path = find_min_transfers(人民广场, 陆家嘴, db)
assert K1路 in path
assert path[-1] == 陆家嘴
db.close()
为什么强调测试？ 因为图算法很容易出边界错误：起点等于终点、无路径、孤立站点。这些场景在print调试中极易遗漏，测试用例能帮你兜底。
优化扩展：从能用到好用
1. 性能优化：缓存热点数据
公交线路和站点是相对静态的，每次查数据库是浪费。用Redis缓存graph结构。
# services/cache.py
import json
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def get_graph_from_cache():
data = r.get('bus_graph')
if data:
return json.loads(data)
return None
def update_graph_cache(graph: dict):
r.set('bus_graph', json.dumps(graph), ex=3600) # 1小时过期
注意：缓存一致性是坑。当线路调整时，必须主动失效缓存。别让用户看到半年前的老线路。
2. 扩展方向：接入真实数据
如果想进阶，可以接入：
OpenStreetMap：获取真实路网和站点坐标。
GTFS数据：谷歌通用 Transit Feed 规范，很多城市开放了公交实时数据。在CSDN上搜“GTFS解析Python”，有大量现成库，如gtfs-kit。
地图API：高德/百度提供POI搜索和路径规划，后端只负责聚合数据。
3. 避坑清单
时区问题：公交时刻表是本地时间，服务器可能在UTC。统一用datetime库处理，存数据库时用UTC，展示时转本地。
站点同名：不同城市可能有同名站点。Station表必须加city_id外键，查询时带上城市参数。
并发查询：BFS在大型城市（上万站点）可能慢。考虑预计算热门OD对（Origin-Destination），或改用Dijkstra（如果考虑时间成本而非仅换乘次数）。
小结
这个项目代码量不到500行，但覆盖了：ORM建模、图论算法、缓存策略、测试驱动。
你不需要一开始就写微服务、K8s、消息队列。先把单体应用跑通，把数据流转逻辑理顺。当你能清晰解释“为什么用BFS而不是Dijkstra”、“为什么加stop_order字段”时，你才算真正入门。
这个知识点你面试被问过吗？ 特别是“如何优化公共交通路径查找算法”或“如何处理动态实时数据与静态路网的冲突”，留言说说你的经历，我挑典型的展开聊聊。 
长河君 · 2026/9/22 8:54:14 · 阅读 1.2k 3天手写实现公交车app，告别看教程不会写的尴尬
3天手写实现公交车app，告别看教程不会写的尴尬
是不是也这样？B站收藏了99+个Python项目，CSDN存了上百篇架构设计，结果真要动手写个公交查询系统，脑子一片空白。卡在“需求拆解”这一步，连数据库表都建不起来。
今天不灌鸡汤，直接带你从零手写实现一个最小可用的公交车app后端。不是那种大而全的演示代码，而是能跑通、能扩展、面试能讲清楚的实战逻辑。跟着敲一遍，你会发现“写项目”没那么玄乎。
项目目标与边界界定
很多新人一上来就想做“高德地图”，结果第一周就崩了。我们得先砍掉90%的需求。
这个手写实现的公交车app，核心功能只有三个：
查询某条线路的所有站点及顺序。
查询某两个站点之间的换乘方案。
获取实时到站预测（模拟数据）。
不做什么同样重要：不做GPS实时定位、不做用户登录鉴权、不做支付、不做地图渲染。这些是前端和第三方API的事，后端只负责数据和逻辑。
为什么这么定？因为初学者最大的坑就是“过度设计”。你在CSDN上搜“公交系统设计”，看到的往往是百万行代码的工业级方案，对你毫无参考价值。我们要的是“最小闭环”，先把数据流转跑通。
目录结构：像搭积木一样组织代码
别把代码全塞在main.py里。哪怕只有500行，也要分层。这是工程化的第一步，也是面试官看重的“代码洁癖”。
bus_app/
├── main.py # 入口，启动FastAPI服务
├── models/ # 数据模型，对应数据库表
│ ├── __init__.py
│ └── bus.py # 线路、站点、班次定义
├── services/ # 业务逻辑层，核心算法在这里
│ ├── __init__.py
│ └── route.py # 换乘路径计算逻辑
├── database/ # 数据库连接与初始化
│ ├── __init__.py
│ ├── db.py # SQLAlchemy连接池
│ └── seed.py # 初始测试数据插入
└── requirements.txt # 依赖管理
关键点：services层不直接操作数据库，它调用models或database提供的方法。这样以后你想把MySQL换成PostgreSQL，只改database层，业务逻辑一行不动。这就是解耦，面试时能讲出这一层，比背八股文有用得多。
核心代码实现：逐行拆解
1. 数据模型：别乱写字段名
很多人用dict存数据，看似灵活，实则致命。结构化数据必须用Pydantic或SQLAlchemy ORM。
# models/bus.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from database.db import Base
class Station(Base):
__tablename__ = 'stations'
id = Column(Integer, primary_key=True)
name = Column(String(50), unique=True, nullable=False)
latitude = Column(Float) # 经纬度用于后续距离计算
longitude = Column(Float)
class Route(Base):
__tablename__ = 'routes'
id = Column(Integer, primary_key=True)
name = Column(String(20), unique=True, nullable=False) # 如 K1路
description = Column(String(200))
class RouteStop(Base):
多对多关系：一条线路有多个站点，一个站点属于多条线路
__tablename__ = 'route_stops'
id = Column(Integer, primary_key=True)
route_id = Column(Integer, ForeignKey('routes.id'))
station_id = Column(Integer, ForeignKey('stations.id'))
stop_order = Column(Integer, nullable=False) # 关键：站点顺序！
route = relationship(Route)
station = relationship(Station)
注意stop_order字段。很多新手会漏掉这个，导致无法判断“下一站”。公交是有序的，不是无序集合。
2. 换乘算法：别用暴力递归
求两点间最短换乘，本质是图的最短路径问题。节点是站点，边是“同一条线路内相邻站点”或“同站换乘”。
# services/route.py
from collections import deque
from models.bus import RouteStop, Station, Route
from database.db import Session
def find_min_transfers(start_name: str, end_name: str, db: Session) - list:
使用BFS寻找最少换乘次数路径
返回: [站点1, 线路1, 站点2, 线路2, ..., 终点]
# 1. 构建邻接表：站点 - [(下一站, 线路名), (换乘线路1), (换乘线路2)...]
graph = {}
stops = db.query(RouteStop).all()
# 初始化所有站点
stations = db.query(Station).all()
for s in stations:
graph[s.name] = []
# 构建同线路相邻关系 + 同站换乘关系
for stop in stops:
# 同线路相邻：当前站 - 下一站
next_stop = db.query(RouteStop).filter_by(
route_id=stop.route_id,
stop_order=stop.stop_order + 1
).first()
if next_stop:
graph[stop.station.name].append((next_stop.station.name, stop.route.name))
# 同站换乘：当前站的其他线路
other_routes = db.query(RouteStop).filter_by(
station_id=stop.station_id
).filter(RouteStop.route_id != stop.route_id).all()
for other in other_routes:
graph[stop.station.name].append((stop.station.name, other.route.name))
# 2. BFS搜索
queue = deque()
queue.append((start_name, [], start_name)) # (当前站, 已走路径, 起始站)
visited = {start_name}
while queue:
current, path, origin = queue.popleft()
if current == end_name:
return path + [current]
for next_station, route_name in graph[current]:
if next_station not in visited:
visited.add(next_station)
# 记录路径：站点 + 经过的线路
new_path = path + [route_name, next_station]
queue.append((next_station, new_path, origin))
return [] # 无路径
逐行解读：
graph 不是简单的站点连站点，而是包含了“通过哪条线路到达”。
BFS保证第一次找到终点时，就是换乘次数最少的路径（假设每站时间成本相同）。
visited 防止死循环，这是图论算法的基本功。
3. 实时到站预测：模拟而非真实
真实公交到站需要GPS数据流，这里我们用“固定间隔+随机误差”模拟。
# services/route.py 补充
import random
def predict_arrival(station_name: str, db: Session) - dict:
模拟到站预测：基于历史平均间隔
假设每条线路每10分钟一班，误差±2分钟
# 实际项目中，这里应查询最近一辆车的GPS位置
# 模拟：随机生成一个1-15分钟的等待时间
wait_time = random.randint(1, 15)
routes = db.query(Route).filter(Route.id.in_(
db.query(RouteStop.route_id).filter_by(
station_id=db.query(Station.id).filter_by(name=station_name).first()
)
)).all()
predictions = []
for r in routes:
predictions.append({
route: r.name,
eta_minutes: wait_time + random.randint(-2, 2),
status: running
})
return predictions
避坑：不要在这里写复杂的机器学习模型预测到站。那是算法工程师的事。后端先保证接口稳定返回结构化的JSON，前端再决定怎么展示。
运行与测试：别等上线才修Bug
1. 初始化测试数据
# database/seed.py
from database.db import SessionLocal
from models.bus import Station, Route, RouteStop
def seed_data():
db = SessionLocal()
# 清空旧数据
db.query(RouteStop).delete()
db.query(Route).delete()
db.query(Station).delete()
# 插入站点
s1 = Station(name=人民广场, latitude=31.23, longitude=121.47)
s2 = Station(name=南京路, latitude=31.24, longitude=121.48)
s3 = Station(name=陆家嘴, latitude=31.25, longitude=121.49)
db.add_all([s1, s2, s3])
db.commit()
# 插入线路
r1 = Route(name=K1路, description=市中心环线)
r2 = Route(name=K2路, description=东西向干线)
db.add_all([r1, r2])
db.commit()
# 插入站点顺序
# K1: 人民广场 - 南京路 - 陆家嘴
# K2: 人民广场 - 陆家嘴 (假设K2也经过，形成换乘)
db.add(RouteStop(route_id=r1.id, station_id=s1.id, stop_order=1))
db.add(RouteStop(route_id=r1.id, station_id=s2.id, stop_order=2))
db.add(RouteStop(route_id=r1.id, station_id=s3.id, stop_order=3))
db.add(RouteStop(route_id=r2.id, station_id=s1.id, stop_order=1))
db.add(RouteStop(route_id=r2.id, station_id=s3.id, stop_order=2))
db.commit()
db.close()
2. 写测试用例
用pytest，不要只靠print调试。
# tests/test_route.py
import pytest
from database.db import SessionLocal
from services.route import find_min_transfers
def test_min_transfers():
db = SessionLocal()
# 从人民广场到陆家嘴，K1直达，应返回K1
path = find_min_transfers(人民广场, 陆家嘴, db)
assert K1路 in path
assert path[-1] == 陆家嘴
db.close()
为什么强调测试？ 因为图算法很容易出边界错误：起点等于终点、无路径、孤立站点。这些场景在print调试中极易遗漏，测试用例能帮你兜底。
优化扩展：从能用到好用
1. 性能优化：缓存热点数据
公交线路和站点是相对静态的，每次查数据库是浪费。用Redis缓存graph结构。
# services/cache.py
import json
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
def get_graph_from_cache():
data = r.get('bus_graph')
if data:
return json.loads(data)
return None
def update_graph_cache(graph: dict):
r.set('bus_graph', json.dumps(graph), ex=3600) # 1小时过期
注意：缓存一致性是坑。当线路调整时，必须主动失效缓存。别让用户看到半年前的老线路。
2. 扩展方向：接入真实数据
如果想进阶，可以接入：
OpenStreetMap：获取真实路网和站点坐标。
GTFS数据：谷歌通用 Transit Feed 规范，很多城市开放了公交实时数据。在CSDN上搜“GTFS解析Python”，有大量现成库，如gtfs-kit。
地图API：高德/百度提供POI搜索和路径规划，后端只负责聚合数据。
3. 避坑清单
时区问题：公交时刻表是本地时间，服务器可能在UTC。统一用datetime库处理，存数据库时用UTC，展示时转本地。
站点同名：不同城市可能有同名站点。Station表必须加city_id外键，查询时带上城市参数。
并发查询：BFS在大型城市（上万站点）可能慢。考虑预计算热门OD对（Origin-Destination），或改用Dijkstra（如果考虑时间成本而非仅换乘次数）。
小结
这个项目代码量不到500行，但覆盖了：ORM建模、图论算法、缓存策略、测试驱动。
你不需要一开始就写微服务、K8s、消息队列。先把单体应用跑通，把数据流转逻辑理顺。当你能清晰解释“为什么用BFS而不是Dijkstra”、“为什么加stop_order字段”时，你才算真正入门。
这个知识点你面试被问过吗？ 特别是“如何优化公共交通路径查找算法”或“如何处理动态实时数据与静态路网的冲突”，留言说说你的经历，我挑典型的展开聊聊。