change handling of background jobs; switch from catalogue/jobs to catalog/results; update and extend tests

Author: pmayd

Changelog.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.5.4
+
+- Change handling of backend jobs: Switch from catalogue/jobs to catalogue/results.
+- Reactivate tests for Regionalstatistik.
+
 ## 0.5.3
 
 - Support for a new parameter in `Table.get_data(..., compress: bool = True)` that can be `True` or `False`. When set to `True`, it will suppress empty rows and columns in the table before downloading it, thus reducing the table size.

README.md

@@ -183,7 +183,7 @@ To contribute to this project, please follow these steps:
 2. Create a new branch by running `git checkout -b <new-branch>` or `git switch -c <new-branch>`. If possible, add an issue number to the branch name.
 3. Make your changes.
 4. Delete the cassettes folder under tests to make sure that the tests are loading the latest data from the API.
-5. Run `uv run pytest tests -sx -vv --vcr-record=new_episodes` to see if all existing tests still run through.
+5. Run `uv run pytest tests -sx -vv --vcr-record=none` to see if all existing tests still run through.
 6. Add new tests depending on your changes.
 7. Run `uv run pytest tests -sx -vv --vcr-record=new_episodes` again to make sure your tests are also passed.
 8. Commit your changes. This will trigger all pre-commit hooks as defined in `.pre-commit-config.yaml`. If any of these pre-hooks fails, your commit is declined, and you have to resolve the issues first.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pystatis"
-version = "0.5.3"
+version = "0.5.4"
 description = "Python wrapper for GENESIS web service interface (API) of the Federal Statistical Office."
 authors = [
     { name = "Michael Aydinbas", email = "[email protected]" },
@@ -73,3 +73,6 @@ log_cli = false
 [tool.safety]
 full_report = true
 ignore = "67599,70612"
+
+[tool.ruff]
+line-length = 100

src/pystatis/__init__.py

@@ -15,7 +15,7 @@
 from pystatis.helloworld import logincheck, whoami
 from pystatis.table import Table
 
-__version__ = "0.5.3"
+__version__ = "0.5.4"
 
 __all__ = [
     "clear_cache",

src/pystatis/config.py

@@ -34,9 +34,7 @@
 REGEX_DB = {
     "genesis": re.compile(r"^((\d{5}-\d{4})|([0-9A-Z]{10}))$"),
     "zensus": re.compile(r"^\d{4}[A-Z]-\d{4}$"),
-    "regio": re.compile(
-        r"^((\d{5}-.{1,2}($|-.*$))|(A.*$)|([0-9A-Z]{10}$)|(\d{5}\w-Z-\d{1,2}))"
-    ),
+    "regio": re.compile(r"^((\d{5}-.{1,2}($|-.*$))|(A.*$)|([0-9A-Z]{10}$)|(\d{5}\w-Z-\d{1,2}))"),
 }
 
 ARS_OR_AGS_MAPPING = {

src/pystatis/db.py

@@ -3,10 +3,8 @@
 import json
 import logging
 
-from pystatis import config
-from pystatis import cache
+from pystatis import cache, config, http_helper
 from pystatis.exception import PystatisConfigError
-from pystatis import http_helper
 
 logger = logging.getLogger(__name__)
 
@@ -34,9 +32,7 @@ def identify_db_matches(table_name: str) -> list[str]:
     if db_matches:
         return db_matches
     else:
-        raise ValueError(
-            f"Could not determine the database for the table '{table_name}'."
-        )
+        raise ValueError(f"Could not determine the database for the table '{table_name}'.")
 
 
 def select_db_by_credentials(db_matches: list[str]) -> str:

src/pystatis/helloworld.py

@@ -20,7 +20,7 @@ def whoami(db_name: str) -> str:
     url = f"{db.get_host(db_name)}" + "helloworld/whoami"
 
     try:
-        response = requests.get(url, timeout=(1, 15))
+        response = requests.get(url, timeout=(30, 15))
     except requests.exceptions.Timeout:
         raise TimeoutError("Login request timed out after 15 minutes")
 
@@ -56,7 +56,7 @@ def logincheck(db_name: str) -> str:
         "language": "de",
     }
 
-    response = requests.post(url, headers=headers, data=params, timeout=(1, 15))
+    response = requests.post(url, headers=headers, data=params, timeout=(30, 15))
 
     # NOTE: Cannot use get_data_from_endpoint due to colliding
     # and misleading usage of "Status" key in API response

src/pystatis/http_helper.py

@@ -7,8 +7,7 @@
 
 import requests
 
-from pystatis import config, db
-from pystatis import cache
+from pystatis import cache, config, db
 from pystatis.exception import DestatisStatusError, NoNewerDataError, TableNotFoundError
 from pystatis.types import ParamDict
 
@@ -51,9 +50,7 @@ def load_data(
             logger.info("Data was loaded from cache.")
         else:
             response = get_data_from_endpoint(endpoint, method, params, db_name)
-            content_type = response.headers.get("Content-Type", "text/csv").split("/")[
-                -1
-            ]
+            content_type = response.headers.get("Content-Type", "text/csv").split("/")[-1]
             data = response.content
 
             # status code 98 means that the table is too big
@@ -62,7 +59,7 @@ def load_data(
             try:
                 # test for job-relevant status code
                 response_status_code = response.json().get("Status").get("Code")
-            except json.decoder.JSONDecodeError:
+            except (json.decoder.JSONDecodeError, requests.exceptions.JSONDecodeError):
                 pass
 
             if response_status_code == 98:
@@ -72,14 +69,9 @@ def load_data(
                     "Verarbeitung im Hintergrund erfolgreich gestartet. Job-ID: %s.",
                     job_id,
                 )
-                # in rare cases it seems that asking catalogue/jobs endpoint for the state of the newly created job fails because no job could be found
-                # so we add 5 seconds here to make sure that the job was created in the meantime
-                time.sleep(5)
-                response = get_data_from_resultfile(job_id, db_name)
+                response = get_data_from_resultfile(job_id, params, db_name)
                 assert isinstance(response.content, bytes)  # nosec assert_used
-                content_type = response.headers.get("Content-Type", "text/csv").split(
-                    "/"
-                )[-1]
+                content_type = response.headers.get("Content-Type", "text/csv").split("/")[-1]
                 data = response.content
 
             cache.cache_data(cache_dir, name, params, data, content_type)
@@ -122,7 +114,7 @@ def get_response(db_name: str, params: ParamDict) -> requests.Response:
             "password": db_pw,
         }
 
-        return requests.post(url, headers=headers, data=params, timeout=(5, 300))
+        return requests.post(url, headers=headers, data=params, timeout=(30, 300))
 
     # Determine database by matching regex to item code
     if db_name is None:
@@ -162,18 +154,19 @@ def start_job(endpoint: str, method: str, params: ParamDict) -> requests.Respons
     Args:
         endpoint (str): Destatis endpoint (eg. data, catalogue, ..)
         method (str): Destatis method (eg. tablefile, ...)
-        params (dict): dictionary of query parameters
+        params (dict): Dictionary of query parameters.
 
     Returns:
         requests.Response: the response object holding the response from calling the Destatis endpoint.
     """
     logger.warning(
         "Die Tabelle ist zu groß, um direkt abgerufen zu werden. Es wird eine Verarbeitung im Hintergrund gestartet."
     )
-    params["job"] = "true"
+    job_params = params.copy()
+    job_params["job"] = "true"
 
     # starting a job
-    response = get_data_from_endpoint(endpoint=endpoint, method=method, params=params)
+    response = get_data_from_endpoint(endpoint=endpoint, method=method, params=job_params)
 
     return response
 
@@ -201,58 +194,57 @@ def get_job_id_from_response(response: requests.Response) -> str:
 
 
 def get_data_from_resultfile(
-    job_id: str, db_name: str | None = None
+    job_id: str, params: ParamDict, db_name: str | None = None
 ) -> requests.Response:
     """Get data from a job once it is finished or when the timeout is reached.
 
     Args:
         job_id (str): Job ID generated by Destatis API.
+        params (dict): Dictionary of query parameters.
         db_name (str, optional): The database to use for this data request.
             One of "genesis", "zensus", "regio". Defaults to None.
 
     Returns:
         requests.Response: the response object holding the response from calling the Destatis endpoint.
     """
-    params = {
-        "selection": "*" + job_id,
-        "searchcriterion": "code",
-        "sortcriterion": "code",
-        "type": "all",
+    job_params = {
+        "selection": job_id,
+        "area": "user",
+        "pagelength": "1",
     }
 
     time_ = time.perf_counter()
 
     while (time.perf_counter() - time_) < JOB_TIMEOUT:
-        response = get_data_from_endpoint(
-            endpoint="catalogue", method="jobs", params=params, db_name=db_name
-        )
-
-        jobs = response.json().get("List")
-        if len(jobs) > 0 and jobs[0].get("State") == "Fertig":
-            logger.info(
-                (
-                    "Verarbeitung im Hintergrund abgeschlossen. "
-                    "Ergebnis kann jetzt abgerufen werden über "
-                    "/data/resultfile und Job-ID: %s."
-                ),
-                job_id,
+        try:
+            response = get_data_from_endpoint(
+                endpoint="catalogue", method="results", params=job_params, db_name=db_name
             )
-            break
-        else:
+
+            jobs = response.json().get("List")
+            if len(jobs) > 0:
+                logger.info(
+                    (
+                        "Verarbeitung im Hintergrund abgeschlossen. "
+                        "Ergebnis kann jetzt abgerufen werden über "
+                        "/data/resultfile und Job-ID: %s."
+                    ),
+                    job_id,
+                )
+                break
+        except DestatisStatusError:
             logger.info("Verarbeitung im Hintergrund läuft noch...")
 
         time.sleep(5)
     else:
-        print("Time out exceeded! Aborting...")
+        logger.error(
+            "Verarbeitungsfenster von %s Minuten überschritten. Job-Datei konnte nicht heruntergeladen werden.",
+            JOB_TIMEOUT // 60,
+        )
         return bytes()
 
-    time.sleep(5)
-    params = {
-        "name": job_id,
-        "area": "all",
-        "compress": "false",
-        "format": "ffcsv",
-    }
+    params = params.copy()
+    params["name"] = job_id
     response = get_data_from_endpoint(
         endpoint="data", method="resultfile", params=params, db_name=db_name
     )
@@ -325,36 +317,47 @@ def _check_destatis_status(destatis_status: dict) -> None:  # type: ignore
     Raises:
         DestatisStatusError: If the status code or type displays an error (caused by the user inputs)
     """
-    # -1 status code for unexpected errors and if no status code is given (faulty response)
-    destatis_status_code = destatis_status.get("Code", -1)
-    destatis_status_type = destatis_status.get("Type", "Information")
+    # Status codes
+    STATUS_CODE_PARAM_ADJUSTED = 22
+    STATUS_CODE_NO_NEW_DATA = 50
+    STATUS_CODE_TABLE_NOT_FOUND = 90
+    STATUS_CODE_LARGE_TABLE = 98
+    STATUS_CODE_NO_MATCHING_OBJECT = 104
+    STATUS_CODE_ERROR = -1  # For unexpected errors and if no status code is given
+
+    # Status types
+    STATUS_TYPE_INFORMATION = "Information"
+    ERROR_TYPES = ["Error", "Fehler"]
+    WARNING_TYPES = ["Warning", "Warnung"]
+
+    destatis_status_code = destatis_status.get("Code", STATUS_CODE_ERROR)
+    destatis_status_type = destatis_status.get("Type", STATUS_TYPE_INFORMATION)
     destatis_status_content = destatis_status.get("Content")
 
-    # define status types
-    error_en_de = ["Error", "Fehler"]
-    warning_en_de = ["Warning", "Warnung"]
-
-    # check for generic/ system error
-    if destatis_status_code == -1:
+    # Check for generic/system error
+    if destatis_status_code == STATUS_CODE_ERROR:
         raise DestatisStatusError(destatis_status_content)
 
-    # check for destatis/ query errors
-    elif (destatis_status_code in [104, 50, 90]) or (
-        destatis_status_type in error_en_de
-    ):
-        if destatis_status_code == 98:
+    # Check for destatis/query errors
+    elif (
+        destatis_status_code
+        in [STATUS_CODE_NO_MATCHING_OBJECT, STATUS_CODE_NO_NEW_DATA, STATUS_CODE_TABLE_NOT_FOUND]
+    ) or (destatis_status_type in ERROR_TYPES):
+        if destatis_status_code == STATUS_CODE_LARGE_TABLE:
             pass
-        elif destatis_status_code == 50:
+        elif destatis_status_code == STATUS_CODE_NO_NEW_DATA:
             raise NoNewerDataError(destatis_status_content)
-        elif destatis_status_code == 90:
+        elif destatis_status_code == STATUS_CODE_TABLE_NOT_FOUND:
             raise TableNotFoundError(destatis_status_content)
         else:
             raise DestatisStatusError(destatis_status_content)
 
-    # output warnings to user
-    elif (destatis_status_code == 22) or (destatis_status_type in warning_en_de):
+    # Output warnings to user
+    elif (destatis_status_code == STATUS_CODE_PARAM_ADJUSTED) or (
+        destatis_status_type in WARNING_TYPES
+    ):
         logger.warning(destatis_status_content)
 
-    # output information to user
-    elif destatis_status_type.lower() == "information":
+    # Output information to user
+    elif destatis_status_type.lower() == STATUS_TYPE_INFORMATION.lower():
         logger.info("Code %d: %s", destatis_status_code, destatis_status_content)

src/pystatis/profile.py

@@ -26,9 +26,7 @@ def change_password(db_name: str, new_password: str) -> str:
     }
 
     # change remote password
-    response_text = load_data(
-        endpoint="profile", method="password", params=params, db_name=db_name
-    )
+    response_text = load_data(endpoint="profile", method="password", params=params, db_name=db_name)
     # change local password
     db.set_pw(db_name, new_password)
 

src/pystatis/results.py

@@ -92,9 +92,7 @@ def show_metadata(self, row_numbers: list[int]) -> None:
                         structure_dict["Head"]["Content"],
                         f"{'-' * 20}",
                         "Columns:",
-                        "\n".join(
-                            [col["Content"] for col in structure_dict["Columns"]]
-                        ),
+                        "\n".join([col["Content"] for col in structure_dict["Columns"]]),
                         f"{'-' * 20}",
                         "Rows:",
                         "\n".join([row["Content"] for row in structure_dict["Rows"]]),