Coverage for src/data_api/api/routers/buses.py: 100%

1""" 

2Buses router. 

3 

4Handles HTTP endpoints for bus-related queries. 

5Delegates business logic to domain services. 

6""" 

7 

8from datetime import datetime 

9from typing import Literal 

10 

11from fastapi import APIRouter, Depends, Query, Response 

12 

13from data_api.api.errors import service_errors 

14from data_api.api.schemas import buses as schemas 

15from data_api.api.schemas.errors import ErrorResponse 

16from data_api.api.schemas.responses import SNAPSHOT_RESPONSES 

17from data_api.domain.buses.query import resolve_bus_time 

18from data_api.runtime import get_runtime 

19 

20router = APIRouter(responses=SNAPSHOT_RESPONSES) 

21 

22 

23async def add_custom_header(response: Response): 

24 """Add X-Data-Commit-Hash header.""" 

25 service = get_runtime().buses 

26 await service.update_data() 

27 if service.last_commit_hash is not None: 

28 response.headers["X-Data-Commit-Hash"] = service.last_commit_hash 

29 

30 

31def get_current_time_state(): 

32 """ 

33 Get current time state (day type and time). 

34 """ 

35 state = resolve_bus_time("current", None, datetime.now()) 

36 return state.day, state.after_time 

37 

38 

39def _query_bus_schedule( 

40 route: schemas.BusRouteType, 

41 day: schemas.BusDayWithCurrent, 

42 direction: schemas.BusDirection, 

43 details: bool, 

44 stop: schemas.BusStopsName | None, 

45 query: schemas.BusQuery, 

46): 

47 if day == "current": 

48 find_day, after_time = get_current_time_state() 

49 else: 

50 state = resolve_bus_time(day, query.time, datetime.now()) 

51 find_day, after_time = state.day, state.after_time 

52 with service_errors(): 

53 return get_runtime().buses.query_schedule( 

54 route_type=route, 

55 day=find_day, 

56 direction=direction, 

57 detailed=details, 

58 stop=stop, 

59 after_time=after_time or "", 

60 limit=query.limit, 

61 ) 

62 

63 

64@router.get( 

65 "/routes", 

66 response_model=list[schemas.BusInfo], 

67 dependencies=[Depends(add_custom_header)], 

68 operation_id="getBusRouteData", 

69 responses={500: {"model": ErrorResponse, "description": "Unable to retrieve bus metadata"}}, 

70) 

71async def get_bus_route_metadata( 

72 route: Literal["main", "nanda"] = Query(None, description="路線選擇"), 

73 direction: Literal["up", "down"] = Query(None, description="方向選擇"), 

74): 

75 """ 

76 取得校園公車資訊。 

77 - 校本部來自[總務處事務組](https://affairs.site.nthu.edu.tw/p/412-1165-20978.php?Lang=zh-tw) 

78 - 南大來自[總務處事務組](https://affairs.site.nthu.edu.tw/p/412-1165-20979.php?Lang=zh-tw) 

79 """ 

80 with service_errors(): 

81 return get_runtime().buses.get_route_info(route, direction) 

82 

83 

84@router.get( 

85 "/stops", 

86 response_model=list[schemas.BusStopsInfo], 

87 dependencies=[Depends(add_custom_header)], 

88 operation_id="getBusStops", 

89 responses={500: {"model": ErrorResponse, "description": "Unable to retrieve bus stops"}}, 

90) 

91async def get_bus_stops_information(): 

92 """取得所有公車站牌的經緯度與資訊。""" 

93 with service_errors(): 

94 return get_runtime().buses.gen_bus_stops_info() 

95 

96 

97@router.get( 

98 "/schedule", 

99 response_model=list[schemas.BusCanonicalDetailedSchedule | schemas.BusCanonicalSchedule], 

100 response_model_exclude_none=True, 

101 dependencies=[Depends(add_custom_header)], 

102 operation_id="getBusSchedule", 

103 responses={500: {"model": ErrorResponse, "description": "Unable to retrieve bus schedules"}}, 

104 response_description="取得公車時刻表資訊。", 

105) 

106async def get_bus_schedule( 

107 route: schemas.BusRouteType = Query(schemas.BusRouteType.all, description="路線選擇"), 

108 day: schemas.BusDayWithCurrent = Query( 

109 schemas.BusDayWithCurrent.current, description="平日、假日或目前時刻" 

110 ), 

111 direction: schemas.BusDirection = Query(schemas.BusDirection.all, description="上山或下山"), 

112 details: bool = Query(False, description="是否包含詳細站點時間資訊"), 

113 stop: schemas.BusStopsName | None = Query( 

114 None, 

115 description=( 

116 "僅回傳停靠此站的班次並於最外層附帶此站的 arrive_time（未指定時省略）；" 

117 "以此站預估到站時間篩選，包含已發車但尚未到站的班次" 

118 ), 

119 ), 

120 query: schemas.BusQuery = Depends(), 

121): 

122 """ 

123 取得指定條件的公車時刻表。 

124 - 不指定條件時，回傳現在開始的全部路線與方向班次，預設最多 5 筆。 

125 - **details=False**: 回傳簡易時刻表（僅發車時間）。 

126 - **details=True**: 回傳詳細時刻表（包含每站預估到達時間）。 

127 - **stop**: 以指定站牌的預估到站時間篩選（包含已發車但尚未到站的班次），並在每筆班次最外層回傳該站預估到達時間 `arrive_time`；未指定 `stop` 時省略 `arrive_time` 並以發車時間篩選。 

128 """ 

129 return _query_bus_schedule(route, day, direction, details, stop, query)