refactor: apply style format and lint, use numpy docstring style

Author: kan-fu

pyproject.toml

@@ -77,6 +77,9 @@ select = [
     "B", # flake8-bugbear
     "SIM", # flake8-simplify
 ]
+ignore = [
+    "E501",
+]
 
 [tool.pytest.ini_options]
 testpaths = [

scratch.ipynb

@@ -19,11 +19,19 @@ class _DataProductFile:
     Is able to poll and wait if required
     """
 
-    def __init__(self, dpRunId: int, index: str, baseUrl: str, token: str, verbosity: str = "INFO"):
+    def __init__(
+        self,
+        dpRunId: int,
+        index: str,
+        baseUrl: str,
+        token: str,
+        verbosity: str = "INFO",
+    ):
         self.verbosity = verbosity
         # Use child logger 'onc.poll' consistent with _PollLog
         from ._Messages import setup_logger
-        self._log = setup_logger('onc.poll', verbosity)
+
+        self._log = setup_logger("onc.poll", verbosity)
         self._retries = 0
         self._status = 202
         self._downloaded = False
@@ -78,7 +86,9 @@ def download(
                         response, outPath, filename, overwrite
                     )
                 except FileExistsError:
-                    self._log.info(f'   Skipping "{self._filePath}": File already exists.')
+                    self._log.info(
+                        f'   Skipping "{self._filePath}": File already exists.'
+                    )
                     self._status = 777
 
             elif self._status == 202:  # Still processing, wait and retry

src/onc/modules/_DataProductFile.py

@@ -1,43 +1,53 @@
 import logging
 import re
-import requests
 import time
 
-REQ_MSG = "Requested: {}"  # get request url
-RESPONSE_TIME_MSG = "Response received in {} seconds." # requests.elapsed value.
-RESPONSE_MSG = "HTTP Response: {} ({})" # Brief description, status code
-MULTIPAGE_MSG = ("The requested data quantity is greater than the "
-                 "supplied row limit and will be downloaded over multiple requests.")
+import requests
 
+REQ_MSG = "Requested: {}"  # get request url
+RESPONSE_TIME_MSG = "Response received in {} seconds."  # requests.elapsed value.
+RESPONSE_MSG = "HTTP Response: {} ({})"  # Brief description, status code
+MULTIPAGE_MSG = (
+    "The requested data quantity is greater than the "
+    "supplied row limit and will be downloaded over multiple requests."
+)
 
 
 LEVEL_MAP = {
-    'CRITICAL': logging.CRITICAL,
-    'ERROR': logging.ERROR,
-    'WARNING': logging.WARNING,
-    'INFO': logging.INFO,
-    'DEBUG': logging.DEBUG,
+    "CRITICAL": logging.CRITICAL,
+    "ERROR": logging.ERROR,
+    "WARNING": logging.WARNING,
+    "INFO": logging.INFO,
+    "DEBUG": logging.DEBUG,
 }
 
 
 class OnclibFormatter(logging.Formatter):
     """
     Custom formatter that removes prefix for INFO level logs.
     """
+
     def format(self, record):
         if record.levelno == logging.INFO:
             return record.getMessage()
         return super().format(record)
 
 
-def setup_logger(logger_name: str = 'onc',
-                 level: int | str = 'INFO') -> logging.Logger:
+def setup_logger(logger_name: str = "onc", level: int | str = "INFO") -> logging.Logger:
     """
     Set up a logger object for displaying verbose messages to console.
 
-    :param logger_name: The unique logger name to use. Can be shared between modules
-    :param level: The logging level to use. Default is 'INFO'.
-    :return: The configured logging.Logger object.
+    Parameters
+    ----------
+    logger_name : str, optional
+        The unique logger name to use. Can be shared between modules.
+    level : int or str, optional
+        The logging level to use. Default is 'INFO'.
+
+    Returns
+    -------
+    logging.Logger
+        The configured logging.Logger object.
     """
 
     # Ensure level is a valid logging level integer
@@ -55,8 +65,10 @@ def setup_logger(logger_name: str = 'onc',
         console.setLevel(level)
 
         # Set the logging format.
-        dtfmt = '%Y-%m-%dT%H:%M:%S'
-        strfmt = f'%(asctime)s.%(msecs)03dZ | %(name)-12s | %(levelname)-8s | %(message)s'
+        dtfmt = "%Y-%m-%dT%H:%M:%S"
+        strfmt = (
+            "%(asctime)s.%(msecs)03dZ | %(name)-12s | %(levelname)-8s | %(message)s"
+        )
         fmt = OnclibFormatter(strfmt, datefmt=dtfmt)
         fmt.converter = time.gmtime
 
@@ -70,15 +82,21 @@ def setup_logger(logger_name: str = 'onc',
     return logger
 
 
-
 def scrub_token(input: str) -> str:
     """
     Replace a token in a query URL or other string with the string 'REDACTED'
     so that users don't accidentally commit their tokens to public repositories
     if ONC Info/Warnings are too verbose.
 
-    :param input: An Oceans 3.0 API URL or string with a token query parameter.
-    :return: A scrubbed url.
+    Parameters
+    ----------
+    input : str
+        An Oceans 3.0 API URL or string with a token query parameter.
+
+    Returns
+    -------
+    str
+        A scrubbed url.
     """
     return re.sub(r"([?&]token=)[a-f0-9-]{36}", r"\1REDACTED", input)
 
@@ -87,9 +105,17 @@ def build_error_message(response: requests.Response, redact_token: bool) -> str:
     """
     Build an error message from a requests.Response object.
 
-    :param response: A requests.Response object.
-    :param redact_token: If true, redact tokens before returning an error message.
-    :return: An error message.
+    Parameters
+    ----------
+    response : requests.Response
+        A requests.Response object.
+    redact_token : bool
+        If true, redact tokens before returning an error message.
+
+    Returns
+    -------
+    str
+        An error message.
     """
     payload = response.json()
     message = payload.get("message")
@@ -98,15 +124,16 @@ def build_error_message(response: requests.Response, redact_token: bool) -> str:
         errors = payload["errors"]
         error_messages = []
         for error in errors:
-            emsg = (f"(API Error Code {error['errorCode']}) "
-                    f"{error['errorMessage']} for query parameter(s) "
-                    f"'{error['parameter']}'.")
+            emsg = (
+                f"(API Error Code {error['errorCode']}) "
+                f"{error['errorMessage']} for query parameter(s) "
+                f"'{error['parameter']}'."
+            )
             error_messages.append(emsg)
-        error_message = '\n'.join(error_messages)
+        error_message = "\n".join(error_messages)
     else:
         error_message = None
-    msg = '\n'.join([m for m in (message, error_message) if m is not None])
-    if redact_token is True and 'token=' in msg:
+    msg = "\n".join([m for m in (message, error_message) if m is not None])
+    if redact_token is True and "token=" in msg:
         msg = scrub_token(msg)
     return msg
-

src/onc/modules/_Messages.py

@@ -4,39 +4,37 @@
 from time import time
 
 import dateutil.parser
-import humanize
+from onc.modules._Messages import (
+    setup_logger,
+)
 
 from ._util import _formatDuration
 
-from onc.modules._Messages import (setup_logger, MULTIPAGE_MSG,
-                                   build_error_message,
-                                   scrub_token,
-                                   REQ_MSG,
-                                   RESPONSE_TIME_MSG,
-                                   RESPONSE_MSG)
-
-
 
 # Handles data multi-page downloads (scalardata, rawdata, archivefiles)
 class _MultiPage:
     def __init__(self, parent: object):
         self.parent = weakref.ref(parent)
         self.result = None
-        self._log = setup_logger('onc.multi', self._config('verbosity') or 'INFO')
+        self._log = setup_logger("onc.multi", self._config("verbosity") or "INFO")
 
     def _config(self, key):
         p = self.parent()
         if p is None:
             return None
-        if hasattr(p, '_config'):
+        if hasattr(p, "_config"):
             return p._config(key)
         return getattr(p, key, None)
 
     def getAllPages(self, service: str, url: str, filters: dict):
         """
         Requests all pages from the service, with the url and filters
         Multiple pages will be downloaded until completed
-        @return: Service response with concatenated data for all pages obtained
+
+        Returns
+        -------
+        dict
+            Service response with concatenated data for all pages obtained.
         """
         # pop archivefiles extension
         extension = None
@@ -48,28 +46,38 @@ def getAllPages(self, service: str, url: str, filters: dict):
         start = time()
         response, responseTime = self._doPageRequest(url, filters, service, extension)
 
-        if isinstance(response,dict):
+        if isinstance(response, dict):
             rNext = response["next"]
 
             if rNext is not None:
-                self._log.info("The requested data quantity is greater than the supplied "
-                                "row limit and will be downloaded over multiple requests.")
+                self._log.info(
+                    "The requested data quantity is greater than the supplied "
+                    "row limit and will be downloaded over multiple requests."
+                )
 
                 pageCount = 1
                 pageEstimate = self._estimatePages(response, service)
                 if pageEstimate > 0:
                     # Exclude the first page when calculating the time estimation
                     timeEstimate = _formatDuration((pageEstimate - 1) * responseTime)
-                    self._log.info(f'Download time for page {pageCount}: {round(responseTime,2)} seconds')
-                    self._log.info(f'Est. number of pages remaining for download: {pageEstimate-1}')
-                    self._log.info(f'Est. number of seconds to download remaining data: {timeEstimate}')
+                    self._log.info(
+                        f"Download time for page {pageCount}: {round(responseTime, 2)} seconds"
+                    )
+                    self._log.info(
+                        f"Est. number of pages remaining for download: {pageEstimate - 1}"
+                    )
+                    self._log.info(
+                        f"Est. number of seconds to download remaining data: {timeEstimate}"
+                    )
 
                 # keep downloading pages until next is None
                 while rNext is not None:
                     pageCount += 1
                     rowCount = self._rowCount(response, service)
 
-                    self._log.info(f"   Submitting request for page {pageCount} ({rowCount} samples)...")
+                    self._log.info(
+                        f"   Submitting request for page {pageCount} ({rowCount} samples)..."
+                    )
 
                     nextResponse, nextTime = self._doPageRequest(
                         url, rNext["parameters"], service, extension
@@ -81,7 +89,9 @@ def getAllPages(self, service: str, url: str, filters: dict):
 
                 totalTime = _formatDuration(time() - start)
 
-                self._log.info(f"   Downloaded {self._rowCount(response, service):d} total samples in {totalTime}.")
+                self._log.info(
+                    f"   Downloaded {self._rowCount(response, service):d} total samples in {totalTime}."
+                )
                 response["next"] = None
 
         return response
@@ -92,8 +102,22 @@ def _doPageRequest(
         """
         Wraps the _doRequest method
         Performs additional processing of the response for certain services
-        @param extension: Only provide for archivefiles filtering
-        Returns a tuple (jsonResponse, duration)
+
+        Parameters
+        ----------
+        url : str
+            API endpoint URL.
+        filters : dict
+            Filters for the request.
+        service : str
+            Name of the service (e.g. archivefile).
+        extension : str, optional
+            Only provide for archivefiles filtering.
+
+        Returns
+        -------
+        tuple
+            (jsonResponse, duration)
         """
         if service.startswith("archivefile"):
             response, duration = self.parent()._doRequest(url, filters, getTime=True)

src/onc/modules/_MultiPage.py

@@ -6,7 +6,7 @@
 
 from ._MultiPage import _MultiPage
 from ._OncService import _OncService
-from ._util import _createErrorMessage, _formatDuration, saveAsFile
+from ._util import _formatDuration, saveAsFile
 
 
 class _OncArchive(_OncService):
@@ -17,7 +17,6 @@ class _OncArchive(_OncService):
     def __init__(self, parent: object):
         super().__init__(parent)
 
-
     def getArchivefileByLocation(self, filters: dict, allPages: bool):
         """
         Return a list of archived files for a device category in a location.
@@ -75,6 +74,7 @@ def downloadArchivefile(self, filename: str = "", overwrite: bool = False):
 
         else:
             from ._Messages import build_error_message
+
             self._log.error(build_error_message(response, self._config("redact_token")))
             response.raise_for_status()