diff --git a/tagoapi/auth.py b/tagoapi/auth.py index 3651029..e69de29 100644 --- a/tagoapi/auth.py +++ b/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 \ No newline at end of file diff --git a/tagoapi/client.py b/tagoapi/client.py index d37ce5d..0d1b5f8 100644 --- a/tagoapi/client.py +++ b/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}") \ No newline at end of file + raise RuntimeError(f"실행중 오류가 발생했습니다. 에러코드: {error_code}") diff --git a/tagoapi/models/ArrivalInfo.py b/tagoapi/models/ArrivalInfo.py index 2cea527..21dfdd0 100644 --- a/tagoapi/models/ArrivalInfo.py +++ b/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"]: diff --git a/tagoapi/models/BaseModel.py b/tagoapi/models/BaseModel.py index 62b4ef9..8d8347a 100644 --- a/tagoapi/models/BaseModel.py +++ b/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 diff --git a/tagoapi/models/Route.py b/tagoapi/models/Route.py index dcfb31f..b8c6a02 100644 --- a/tagoapi/models/Route.py +++ b/tagoapi/models/Route.py @@ -63,6 +63,26 @@ def from_dict(cls, data: dict) -> "Route": intervalsattime=data.get("intervalsattime"), intervalsuntime=data.get("intervalsuntime") ) + + @property + def route_id(self) -> str: + return self.routeId + + @property + def route_no(self) -> str: + return self.routeNo + + @property + def route_type(self) -> str: + return self.routeTp + + @property + def end_node_name(self) -> str: + return self.endNodeNm + + @property + def start_node_name(self) -> str: + return self.startNodeNm # @classmethod # def from_list(cls, data: list[dict]) -> list["Route"]: diff --git a/tagoapi/models/Station.py b/tagoapi/models/Station.py index 8107551..255b928 100644 --- a/tagoapi/models/Station.py +++ b/tagoapi/models/Station.py @@ -49,8 +49,40 @@ def from_dict(cls, data: dict) -> "Station": updowncd = data.get("updowncd"), nodeord = data.get("nodeord"), ) + + @property + def node_id(self) -> str: + return self.nodeId + + @property + def station_id(self) -> str: + return self.nodeId + + @property + def node_name(self) -> str: + return self.nodeNm + + @property + def station_name(self) -> str: + return self.nodeNm + + @property + def node_no(self) -> int: + return self.nodeNo + + @property + def station_no(self) -> int: + return self.nodeNo + + @property + def gps_lati(self) -> float: + return self.gpsLati + + @property + def gps_long(self) -> float: + return self.gpsLong # @classmethod # def from_list(cls, data: list[dict]) -> list["Station"]: # return [cls.from_dict(station, client) for station in data] - \ No newline at end of file + diff --git a/tagoapi/models/Vehicle.py b/tagoapi/models/Vehicle.py index 9f987f6..d4ab2fe 100644 --- a/tagoapi/models/Vehicle.py +++ b/tagoapi/models/Vehicle.py @@ -16,6 +16,7 @@ def __init__( route: Route, gpsLati: float = None, gpsLong: float = None, + nodeNo: int = None, arrtime: int = None, arrprevstationcnt: int = None, vehicleTp: str = None, @@ -25,6 +26,7 @@ def __init__( self.route = route self.gpsLati = gpsLati self.gpsLong = gpsLong + self.nodeNo = nodeNo self.arrtime = arrtime self.arrprevstationcnt = arrprevstationcnt self.vehicleTp = vehicleTp @@ -42,12 +44,29 @@ def from_dict(cls, data: dict) -> "Vehicle": route=Route.from_dict(data), gpsLati=data.get("gpslati"), gpsLong=data.get("gpslong"), + nodeNo=data.get("nodeno"), arrtime=data.get("arrtime"), arrprevstationcnt=data.get("arrprevstationcnt"), vehicleTp=data.get("vehicletp"), vehicleNo=data.get("vehicleno") ) + + @property + def route_id(self) -> str: + return self.route.routeId + + @property + def station_no(self) -> int: + return self.nodeNo + + @property + def gps_lati(self) -> float: + return self.gpsLati + + @property + def gps_long(self) -> float: + return self.gpsLong # @classmethod # def from_list(cls, data: list[dict]) -> list["Vehicle"]: - # return [ cls.from_dict(vehicle) for vehicle in data] \ No newline at end of file + # return [ cls.from_dict(vehicle) for vehicle in data] diff --git a/tagoapi/utils/decorator/convert_model.py b/tagoapi/utils/decorator/convert_model.py index 2815cf9..a857ee2 100644 --- a/tagoapi/utils/decorator/convert_model.py +++ b/tagoapi/utils/decorator/convert_model.py @@ -19,6 +19,7 @@ def convert_model( is_list: bool = True ): # 데코레이터가 사용할 매개변수 def decorator(fn): # 호출할 함수를 매개변수로 받음 + @wraps(fn) def inner(self, *args, **kwargs): # 호출할 함수의 매개변수를 받아서 이를 실행 key = _generate_cache_key(*args, _fname=fn.__name__, **kwargs) if is_cached else None cached = cache.get(key) if key else None @@ -32,11 +33,12 @@ def inner(self, *args, **kwargs): # 호출할 함수의 매개변수를 받아 else: raw = cached if model: + city_code = raw.get("city_code", raw.get("cityCode", raw.get("citycode"))) ## convert list if isinstance(raw.get("result"), list): - res = model.from_list(raw.get("result"), raw.get("citycode")) + res = model.from_list(raw.get("result"), city_code=city_code) else: - res = model.from_dict({**(raw.get("result")), "citycode": raw.get("cityCode")}) + res = model.from_dict({**(raw.get("result")), "citycode": city_code}) res.set_client(self) return BaseList([res]) if is_list and not isinstance(res, BaseList) else res @@ -45,4 +47,3 @@ def inner(self, *args, **kwargs): # 호출할 함수의 매개변수를 받아 return raw return inner return decorator - diff --git a/tagoapi/utils/params.py b/tagoapi/utils/params.py index 5d09fb5..78150a7 100644 --- a/tagoapi/utils/params.py +++ b/tagoapi/utils/params.py @@ -1,36 +1,37 @@ -from typing import TYPE_CHECKING -from typing import TypeVar +from typing import Union, Optional, List -U = TypeVar('U', dict[str], list) # list, dict으로 반환할 때 - - -if TYPE_CHECKING: - from tagoapi import TAGOAuth def build_params( - auth: "TAGOAuth", - numOfRows: int = 300, - pageNo: int = 1, - **kwargs: dict - ) -> dict: + service_key: str, + num_of_rows: int = 300, + page_no: int = 1, + kwargs: dict = None +) -> dict: + #TODO: 설명 수정 + """ + 요청을 보내기 위한 파라미터를 완성합니다. + :param service_key: 서비스 키 + :param num_of_rows: 항목당 나올 횟수 설정 + :param page_no: 페이지 + :param kwargs: 안에 들어가야하는 파라미터 + """ return { - "serviceKey": auth.serviceKey, - "numOfRows": numOfRows, - "pageNo": pageNo, + "serviceKey": service_key, + "numOfRows": num_of_rows, + "pageNo": page_no, "_type": "json", - **{key: value for key, value in kwargs.items() if value} + **kwargs, } -def parse_metadata(res: dict) -> U | None: +def parse_metadata(res: dict) -> Optional[dict]: striped = res.get("response", {}).get("body", {}).get("items", {}) if isinstance(striped, dict): return striped.get("item", None) - return None -def _check_bracket(data: str) -> list[str]: +def _check_bracket(data: str) -> List[str]: args = [] current_match = None fa = '' @@ -55,16 +56,12 @@ def __init__(self, raw_key): self.raw_key = raw_key self._args = _check_bracket(self.raw_key) - @property def key_args(self): return self._args def generate_key(self, data: dict) -> str: generated_key = self.raw_key - # if len(kwargs) > len(self._args): - # raise TypeError(f"generate_key() takes {len(self._args)} positional argument but {len(kwargs)} were given") - # TODO: 이거 self._args말고 kwargs.keys해서 arg랑 대응 시켜서 없으면 raise 이런식으로 작성해도 좋을 듯 for arg in self._args: k = data.get(arg.lower(), None)