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 Baseclass 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 Sessiondef 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 randomdef 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, RouteStopdef 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_transfersdef test_min_transfers():db = SessionLocal()# 从人民广场到陆家嘴，K1直达，应返回K1path = find_min_transfers(人民广场, 陆家嘴, db)assert K1路 in pathassert path[-1] == 陆家嘴db.close()为什么强调测试？ 因为图算法很容易出边界错误：起点等于终点、无路径、孤立站点。这些场景在print调试中极易遗漏，测试用例能帮你兜底。
优化扩展：从能用到好用
1. 性能优化：缓存热点数据
公交线路和站点是相对静态的，每次查数据库是浪费。用Redis缓存graph结构。
# services/cache.py
import json
import redisr = 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 Nonedef 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字段”时，你才算真正入门。
这个知识点你面试被问过吗？ 特别是“如何优化公共交通路径查找算法”或“如何处理动态实时数据与静态路网的冲突”，留言说说你的经历，我挑典型的展开聊聊。
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 Baseclass 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 Sessiondef 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 randomdef 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, RouteStopdef 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_transfersdef test_min_transfers():db = SessionLocal()# 从人民广场到陆家嘴，K1直达，应返回K1path = find_min_transfers(人民广场, 陆家嘴, db)assert K1路 in pathassert path[-1] == 陆家嘴db.close()为什么强调测试？ 因为图算法很容易出边界错误：起点等于终点、无路径、孤立站点。这些场景在print调试中极易遗漏，测试用例能帮你兜底。
优化扩展：从能用到好用
1. 性能优化：缓存热点数据
公交线路和站点是相对静态的，每次查数据库是浪费。用Redis缓存graph结构。
# services/cache.py
import json
import redisr = 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 Nonedef 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字段”时，你才算真正入门。
这个知识点你面试被问过吗？ 特别是“如何优化公共交通路径查找算法”或“如何处理动态实时数据与静态路网的冲突”，留言说说你的经历，我挑典型的展开聊聊。
