From 140b19da18ae2d59cb771ac07be329a4c3539735 Mon Sep 17 00:00:00 2001 From: Guillaume Latour Date: Tue, 9 Jul 2024 10:28:03 +0200 Subject: [PATCH] fix: download non zipped report if zip does not exist --- pyproject.toml | 2 +- .../traffic_stats/models/responses.py | 75 ++++++++++++------- 2 files changed, 51 insertions(+), 26 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index db7596b..196fe62 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,7 @@ maintainers = [ name = "tomtom_api" readme = "readme.md" requires-python = ">=3.8" -version = "2024.7.4" +version = "2024.7.9" [project.urls] repository = "https://stash.macq.eu/projects/RDDS/repos/tomtom-api" diff --git a/src/tomtom_api/traffic_stats/models/responses.py b/src/tomtom_api/traffic_stats/models/responses.py index 71dbd23..b724962 100644 --- a/src/tomtom_api/traffic_stats/models/responses.py +++ b/src/tomtom_api/traffic_stats/models/responses.py @@ -2,6 +2,7 @@ In this module are grouped all the dataclasses that are hydrated when the API is responding something. """ + from __future__ import annotations import datetime as dt @@ -47,7 +48,13 @@ class TomtomResponseStatus(JSONWizard, LoadMixin): response_status: str urls: Optional[List[str]] - def __init__(self, job_id: int, job_state: TomtomJobState, response_status: str, urls: Optional[List[str]] = None): + def __init__( + self, + job_id: int, + job_state: TomtomJobState, + response_status: str, + urls: Optional[List[str]] = None, + ): self.job_id = job_id self.job_state = job_state self.response_status = response_status @@ -57,7 +64,7 @@ def load_to_enum(o, base_type: Type[Enum]) -> Enum: return base_type.from_str(o) def display_info(self) -> str: - return f'{self.job_id} ({self.job_state.name})' + return f"{self.job_id} ({self.job_state.name})" def download(self, file_type: TomtomDownloadFileType) -> bytes: """Download the job report, in the given format. @@ -82,24 +89,38 @@ def download(self, file_type: TomtomDownloadFileType) -> bytes: This exception can also be triggered if there is an error while downloading the file. """ if self.job_state != TomtomJobState.DONE: - raise DownloadException(f'The "DONE" state is required for downloading the report ({self.job_state}).') + raise DownloadException( + f'The "DONE" state is required for downloading the report ({self.job_state}).' + ) if self.urls is None: - raise DownloadException('There is no download url for this job.') + raise DownloadException("There is no download url for this job.") + + filtered_links = [link for link in self.urls if f"{file_type.value}?" in link] + + # if nothing is found, remove the compression extension + if len(filtered_links) == 0: + filtered_links = [ + link + for link in self.urls + if f"{file_type.value.replace('.zip', '')}?" in link + ] - filtered_links = [link for link in self.urls if f'{file_type.value}?' in link] + # then perform a sanity check if len(filtered_links) != 1: - log.debug(f'Available links: {self.urls}.') - log.debug(f'Provided file type: {file_type}.') - raise DownloadException(f'Impossible to find the desired file in the following links: {filtered_links}.') + log.debug(f"Available links: {self.urls}.") + log.debug(f"Provided file type: {file_type}.") + raise DownloadException( + f"Impossible to find the desired file in the following links: {filtered_links}." + ) link = filtered_links[0] - log.debug(f'Performing request to {link}') + log.debug(f"Performing request to {link}") response = requests.get(link) if response.status_code != 200: - log.debug(f'Response: {response}') - raise DownloadException(f'Error while downloading {link}') + log.debug(f"Response: {response}") + raise DownloadException(f"Error while downloading {link}") return response.content @@ -124,10 +145,10 @@ def write(self, file: Path, file_type: TomtomDownloadFileType) -> None: """ content = self.download(file_type) if file.exists(): - log.warning(f'The following file will be overwritten: {file}') + log.warning(f"The following file will be overwritten: {file}") - log.debug(f'Writing file {file}') - with open(file, 'wb') as f: + log.debug(f"Writing file {file}") + with open(file, "wb") as f: f.write(content) @@ -145,25 +166,29 @@ class TomtomJobInfo(JSONWizard, LoadMixin): name: str created_at: dt.datetime state: TomtomJobState - job_id: int = json_field('id', all=True) - job_type: str = json_field('type', all=True) - completed_at: Optional[dt.datetime] = json_field('completed_at', default=None) + job_id: int = json_field("id", all=True) + job_type: str = json_field("type", all=True) + completed_at: Optional[dt.datetime] = json_field("completed_at", default=None) def load_to_enum(o, base_type: Type[Enum]) -> Enum: return base_type.from_str(o) def display_info(self) -> str: - datetime_fmt = '%Y-%m-%d %H:%M:%S' + datetime_fmt = "%Y-%m-%d %H:%M:%S" start = self.created_at.strftime(datetime_fmt) - end = 'None' if self.completed_at is None else self.completed_at.strftime(datetime_fmt) - return f'┌─ [{self.job_id}] {self.name} // {self.state.name}\n└─ {self.job_type} <{start} ⟶ {end}>' + end = ( + "None" + if self.completed_at is None + else self.completed_at.strftime(datetime_fmt) + ) + return f"┌─ [{self.job_id}] {self.name} // {self.state.name}\n└─ {self.job_type} <{start} ⟶ {end}>" @dataclass class Sort(JSONWizard): - is_sorted: bool = json_field('sorted', all=True) - is_unsorted: bool = json_field('unsorted', all=True) - is_empty: bool = json_field('empty', all=True) + is_sorted: bool = json_field("sorted", all=True) + is_unsorted: bool = json_field("unsorted", all=True) + is_empty: bool = json_field("empty", all=True) @dataclass @@ -194,5 +219,5 @@ class TomtomResponseSearchJobs(JSONWizard): @dataclass class JsonIpResponse(JSONWizard): ip: str - geo_ip: str = json_field('geo-ip', all=True) - api_help: str = json_field('API Help', all=True) + geo_ip: str = json_field("geo-ip", all=True) + api_help: str = json_field("API Help", all=True)