Refactor namepsace

tagoapi/auth.py

@@ -1,12 +0,0 @@
-
-
-class TAGOAuth:
-    def __init__(self, serviceKey: str):
-        if not isinstance(serviceKey, str):
-            raise TypeError("Service key must be a string")
-
-        self._serviceKey = serviceKey
-
-    @property
-    def serviceKey(self):
-        return self._serviceKey

tagoapi/client.py

@@ -2,7 +2,6 @@
 from .utils.decorator import *
 from .utils import *
 from .models import *
-from .auth import TAGOAuth
 
 from typing import Optional, overload
 
@@ -17,161 +16,220 @@ class TAGOClient:
     CACHE_TTL = 604800
 
 
-    def __init__(self, auth: TAGOAuth):
-        if not isinstance(auth, TAGOAuth):
-            raise TypeError("Expected 'auth' to be an instance of TAGOAuth")
-
-        self.auth = auth
+    def __init__(self, service_key: str):
+        self.service_key = service_key
 
     @overload
-    def get_station(self, cityCode: int, nodeNo: int) -> list[Station]: ...
+    def get_station(self, city_code: int, station_no: int) -> list[Station]: ...
     @overload
-    def get_station(self, cityCode: int, nodeNo: Optional[int], nodeNm: str) -> list[Station]: ...
+    def get_station(self, city_code: int, station_no: Optional[int], station_name: str) -> list[Station]: ...
 
     @convert_model(604800, Route)
     def get_route_by_no(
         self,
-        cityCode: int,
-        routeNo: str
+        city_code: int = None,
+        route_no: str = None,
+        cityCode: int = None,
+        routeNo: str = None,
     ) -> list[Route]:
-        """노선 번호로 버스를 조회합니다"""
+        """노선 번호로 버스를 조회합니다
+        :param city_code: 도시 코드
+        :param route_no: 노선 번호
+        """
+        city_code = city_code if city_code is not None else cityCode
+        route_no = route_no if route_no is not None else routeNo
         endpoint = f'{self.BUSROUTE}/getRouteNoList'
-        params = build_params(self.auth, cityCode=cityCode, routeNo=routeNo)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "route_no": route_no})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(604800, Route, is_list=False)
     def get_route_by_id(
         self,
-        cityCode: int,
-        routeId: str
+        city_code: int = None,
+        route_id: str = None,
+        cityCode: int = None,
+        routeId: str = None,
     ) -> Route:
         """노선 ID로 버스 정보를 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        route_id = route_id if route_id is not None else routeId
         endpoint = f'{self.BUSROUTE}/getRouteInfoIem'
-        params = build_params(self.auth, cityCode=cityCode, routeId=routeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "route_id": route_id})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(604800, Route)
     def get_route_by_station(
         self,
-        cityCode: int,
-        nodeId: str
+        city_code: int = None,
+        station_id: str = None,
+        cityCode: int = None,
+        nodeId: str = None,
     ) -> list[Route]:
         """정류소를 경유하는 노선을 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        station_id = station_id if station_id is not None else nodeId
         endpoint = f'{self.BUSTATION}/getSttnThrghRouteList'
-        params = build_params(self.auth, cityCode=cityCode, nodeid=nodeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "station_id": station_id})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
 
     @convert_model(604800, Station)
     def get_station_by_route(
         self,
-        cityCode: int,
-        routeId: str
+        city_code: int = None,
+        route_id: str = None,
+        cityCode: int = None,
+        routeId: str = None,
     ) -> list[Station]:
         """노선이 경유하는 정류소를 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        route_id = route_id if route_id is not None else routeId
         endpoint = f'{self.BUSROUTE}/getRouteAcctoThrghSttnList'
-        params= build_params(self.auth, cityCode=cityCode, routeId=routeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "route_id": route_id})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(86400, Station)
     def get_station(
         self,
-        cityCode: int,
+        city_code: int = None,
+        station_no: int = None,
+        station_name: str = None,
+        cityCode: int = None,
         nodeNo: int = None,
         nodeNm: str = None,
     ) -> list[Station]:
         """정류소명 또는 번호로 정류소를 조회합니다"""
-        if not (nodeNo or nodeNm):
-            raise ValueError("Only one of 'nodeNo' or 'nodeNm' should be provided.")
+        city_code = city_code if city_code is not None else cityCode
+        station_no = station_no if station_no is not None else nodeNo
+        station_name = station_name if station_name is not None else nodeNm
+        if not (station_no or station_name):
+            raise ValueError("Only one of 'station_no' or 'station_name' should be provided.")
 
         endpoint = f'{self.BUSTATION}/getSttnNoList'
-        params= build_params(self.auth, cityCode=cityCode, nodeNm=nodeNm,nodeNo=nodeNo)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "station_no": station_no, "station_name": station_name})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(86400, Station, is_cached=False)
     def get_station_by_gps(
         self,
-        gpsLati: float,
-        gpsLong: float,
+        gps_lati: float = None,
+        gps_long: float = None,
+        gpsLati: float = None,
+        gpsLong: float = None,
     ) -> list[Station]:
         """GPS 좌표 기반으로 주변 정류소를 조회합니다"""
+        gps_lati = gps_lati if gps_lati is not None else gpsLati
+        gps_long = gps_long if gps_long is not None else gpsLong
         endpoint = f'{self.BUSTATION}/getCrdntPrxmtSttnList'
-        params = build_params(self.auth, gpsLati=gpsLati, gpsLong=gpsLong)
+        params = build_params(self.service_key, kwargs={"gps_lati": gps_lati, "gps_long": gps_long})
         return self._fetch_and_convert(endpoint, params)
 
     @convert_model(model=ArrivalInfo, is_cached=False)
     def get_arrival_by_station(
         self,
-        cityCode: int,
-        nodeId: str,
+        city_code: int = None,
+        station_id: str = None,
+        node_id: str = None,
+        cityCode: int = None,
+        nodeId: str = None,
     ) -> list[ArrivalInfo]:
         """실시간 도착예정정보 및 운행정보 목록을 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        normalized_station_id = station_id or node_id or nodeId
         endpoint = f'{self.AVRINFO}/getSttnAcctoArvlPrearngeInfoList'
-        params = build_params(self.auth, cityCode=cityCode, nodeId=nodeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "nodeId": normalized_station_id})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(model=ArrivalInfo, is_cached=False)
     def get_route_arrival_by_station(
         self,
-        cityCode: int,
-        nodeId: str,
-        routeId: str,
+        city_code: int = None,
+        station_id: str = None,
+        node_id: str = None,
+        route_id: str = None,
+        cityCode: int = None,
+        nodeId: str = None,
+        routeId: str = None,
     ) -> list[ArrivalInfo]:
         """특정노선의 실시간 도착예정정보 및 운행정보 목록을 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        normalized_station_id = station_id or node_id or nodeId
+        normalized_route_id = route_id or routeId
         endpoint = f'{self.AVRINFO}/getSttnAcctoSpcifyRouteBusArvlPrearngeInfoList'
-        params = build_params(self.auth, cityCode=cityCode, nodeId=nodeId, routeId=routeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(
+            self.service_key,
+            kwargs={"city_code": city_code, "nodeId": normalized_station_id, "routeId": normalized_route_id},
+        )
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(model=Vehicle, is_cached=False)
     def get_route_pos(
         self, 
-        cityCode: int,
-        routeId: int,
+        city_code: int = None,
+        route_id: int = None,
+        cityCode: int = None,
+        routeId: int = None,
     ) -> list[Vehicle]:
         """버스의 S위치정보의 목록을 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        normalized_route_id = route_id if route_id is not None else routeId
         endpoint = f'{self.BUSPOS}/getRouteAcctoBusLcList'
-        params = build_params(self.auth, cityCode=cityCode, routeId=routeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(self.service_key, kwargs={"city_code": city_code, "routeId": normalized_route_id})
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
     @convert_model(model=Vehicle, is_cached=False)
     def get_route_pos_near_station(
         self, 
-        cityCode: int,
-        routeId: int,
-        nodeId: int,
+        city_code: int = None,
+        route_id: int = None,
+        station_id: int = None,
+        node_id: int = None,
+        cityCode: int = None,
+        routeId: int = None,
+        nodeId: int = None,
     ) -> list[Vehicle]:
         """특정정류소에 접근한 버스의 위치정보를 조회합니다"""
+        city_code = city_code if city_code is not None else cityCode
+        normalized_route_id = route_id if route_id is not None else routeId
+        normalized_station_id = (
+            station_id
+            if station_id is not None
+            else (node_id if node_id is not None else nodeId)
+        )
         endpoint = f'{self.BUSPOS}/getRouteAcctoSpcifySttnAccesBusLcInfo'
-        params = build_params(self.auth, cityCode=cityCode, routeId=routeId, nodeId=nodeId)
-        return self._fetch_and_convert(endpoint, params, citycode=cityCode)
+        params = build_params(
+            self.service_key,
+            kwargs={"city_code": city_code, "routeId": normalized_route_id, "nodeId": normalized_station_id},
+        )
+        return self._fetch_and_convert(endpoint, params, city_code=city_code)
 
 
     ######## method for LazyLoading ################
 
 
     def _get_route(self, route: Route) -> Route:
-        return self.get_route_by_id(route.cityCode, route.routeId)
+        return self.get_route_by_id(route.city_code, route.route_id)
 
     def _get_stations_by_route(self, route: Route) -> list[Station]:
-        return self.get_station_by_route(route.cityCode, route.routeId)
+        return self.get_station_by_route(route.city_code, route.route_id)
 
     def _get_station(self, station: Station) -> Station:
-        return self.get_station(station.cityCode, nodeNm=station.nodeNm)[0]
+        return self.get_station(station.city_code, station_name=station.station_name)[0]
 
     def _get_routes_by_station(self, station: Station) -> list[Route]:
-        return self.get_route_by_station(station.cityCode, station.nodeId)
+        return self.get_route_by_station(station.city_code, station.station_id)
 
-    def _get_station_by_arrival_info(self, arrivalInfo: ArrivalInfo) -> Station:
-        return self.get_station(arrivalInfo.cityCode, nodeNo=arrivalInfo.nodeNo)[0]
+    def _get_station_by_arrival_info(self, arrival_info: ArrivalInfo) -> Station:
+        return self.get_station(arrival_info.city_code, station_no=arrival_info.station_no)[0]
 
-    def _get_route_by_arrival_info(self, arrivalInfo: ArrivalInfo) -> Route:
-        return self.get_route_by_id(arrivalInfo.cityCode, arrivalInfo.routeId)
+    def _get_route_by_arrival_info(self, arrival_info: ArrivalInfo) -> Route:
+        return self.get_route_by_id(arrival_info.city_code, arrival_info.route_id)
 
     def _get_route_by_vehicle(self, vehicle: Vehicle) -> Route:
-        return self.get_route_by_id(vehicle.cityCode, vehicle.routeId)
+        return self.get_route_by_id(vehicle.city_code, vehicle.route_id)
 
     def _get_station_by_vehicle(self, vehicle: Vehicle) -> Station:
-        return self.get_station(vehicle.cityCode, nodeNo=vehicle.nodeNo)[0];
+        return self.get_station(vehicle.city_code, station_no=vehicle.station_no)[0];
 
     ######## get util ################
 
@@ -180,11 +238,11 @@ def _fetch_and_convert(
             self,
             endpoint: str,
             params: dict,
-            citycode: int,
+            city_code: int,
             **kwargs
     ) -> list | dict:
         response = parse_metadata(self._get(endpoint, params))
-        return {"result":response, "cityCode": citycode}
+        return {"result":response, "city_code": city_code}
 
     def _get(self, endpoint: str, params: dict) -> any:
         response = http_get(f"{self.BASE_URL}/{endpoint}", params=params)
@@ -204,4 +262,4 @@ def _get(self, endpoint: str, params: dict) -> any:
         elif error_code == '32':
             raise UnRegisteredIpError("등록되지 않은 IP입니다.")
         else:
-            raise RuntimeError(f"실행중 오류가 발생했습니다. 에러코드: {error_code}")
+            raise RuntimeError(f"실행중 오류가 발생했습니다. 에러코드: {error_code}")

tagoapi/models/ArrivalInfo.py

@@ -1,4 +1,6 @@
 from .BaseModel import BaseModel
+from .Station import Station
+from .Route import Route
 from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from .Station import Station
@@ -41,6 +43,22 @@ def from_dict(cls, data: dict) -> "ArrivalInfo":
             vehicleTp = data.get("vehicletp"),
             arrtime = data.get("arrtime")
         )
+
+    @property
+    def node_id(self) -> str:
+        return self.node.nodeId
+
+    @property
+    def station_id(self) -> str:
+        return self.node.nodeId
+
+    @property
+    def station_no(self) -> int:
+        return self.node.nodeNo
+
+    @property
+    def route_id(self) -> str:
+        return self.route.routeId
 
     # @classmethod
     # def from_list(cls, data: list[dict]) -> list["ArrivalInfo"]:

tagoapi/models/BaseModel.py

@@ -10,9 +10,9 @@ class BaseModel:
     _key = KeyExtract(cache_key)
     _lazy_fields: dict = {}
 
-    def __init__(self, cityCode: int):
+    def __init__(self, city_code: int):
         self._client = None
-        self.cityCode = cityCode
+        self.city_code = city_code
 
     def to_dict(self) -> dict:
         return vars(self)
@@ -24,8 +24,17 @@ def set_client(self, client: "TAGOClient"):
     def from_dict(cls, data: dict) -> "BaseModel": ...
 
     @classmethod
-    def from_list(cls, data: list, cityCode: int) -> BaseList:
-        return BaseList([cls.from_dict({**d, "cityCode": cityCode}) for d in data])
+    def from_list(
+        cls,
+        data: list,
+        city_code: int = None,
+        cityCode: int = None,
+    ) -> BaseList:
+        normalized_city_code = city_code if city_code is not None else cityCode
+        return BaseList(
+            cls.from_dict({**d, "citycode": normalized_city_code})
+            for d in data
+        )
 
     def __getattr__(self, item):
 
@@ -68,3 +77,10 @@ def __getattribute__(self, item):
     def key(self) -> KeyExtract:
         return self._key
 
+    @property
+    def cityCode(self) -> int:
+        return self.city_code
+
+    @cityCode.setter
+    def cityCode(self, value: int):
+        self.city_code = value