Replace cgi and logging.warn with new APIs (#292)

The `cgi` Python module has been deprecated in PEP 594. In particular,
the `parse_header` function has been deprecated in favor of the new
`email.message.Message` API. This commit replaces all occurrences of
`parse_header` with the new API. This commit also adds a test for the
`remotepath.download` API to avoid regressions.

As of Python 3.3, the `logging.warn` API has been deprected in favor of
the `logging.warning` method. This commit substitutes all occurrences of
the `warn()` method with the equivalent `warning()` API.

streamflow/config/config.py

@@ -42,7 +42,7 @@ def __init__(self, name: str, config: MutableMapping[str, Any]) -> None:
             self.deplyoments = config.get("models", {})
             if self.deplyoments:
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         "The `models` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                         "Use `deployments` instead."
                     )

streamflow/cwl/translator.py

@@ -1533,7 +1533,7 @@ def _translate_command_line_tool(
                 # Otherwise, throw a warning and skip the DockerRequirement conversion
                 else:
                     if logger.isEnabledFor(logging.WARN):
-                        logger.warn(
+                        logger.warning(
                             f"Skipping DockerRequirement conversion for step `{name_prefix}` "
                             f"when executing on `{target.deployment.name}` deployment."
                         )

streamflow/data/remotepath.py

@@ -1,16 +1,17 @@
 from __future__ import annotations
 
 import asyncio
-import cgi
 import errno
 import glob
 import hashlib
 import os
 import posixpath
 import shutil
+from email.message import Message
 from typing import MutableSequence, TYPE_CHECKING
 
 import aiohttp
+from aiohttp import ClientResponse
 
 from streamflow.core import utils
 from streamflow.core.context import StreamFlowContext
@@ -41,6 +42,15 @@ def _file_checksum_local(path: str) -> str:
         return sha1_checksum.hexdigest()
 
 
+def _get_filename_from_response(response: ClientResponse, url: str):
+    if cd_header := response.headers.get("Content-Disposition"):
+        message = Message()
+        message["Content-Disposition"] = cd_header
+        if filename := message.get_param("filename"):
+            return filename
+    return url.rsplit("/", 1)[-1]
+
+
 def _listdir_local(path: str, file_type: FileType | None) -> MutableSequence[str]:
     content = []
     dir_content = os.listdir(path)
@@ -95,25 +105,22 @@ async def download(
         async with aiohttp.ClientSession() as session:
             async with session.get(url) as response:
                 if response.status == 200:
-                    _, params = cgi.parse_header(
-                        response.headers.get("Content-Disposition", "")
+                    filepath = os.path.join(
+                        parent_dir, _get_filename_from_response(response, url)
                     )
-                    filepath = os.path.join(parent_dir, params["filename"])
-                    content = await response.read()
+                    with open(filepath, mode="wb") as f:
+                        f.write(await response.read())
                 else:
                     raise Exception(
                         f"Downloading {url} failed with status {response.status}:\n{response.content}"
                     )
-        with open(filepath, mode="wb") as f:
-            f.write(content)
     else:
         async with aiohttp.ClientSession() as session:
             async with session.head(url, allow_redirects=True) as response:
                 if response.status == 200:
-                    _, params = cgi.parse_header(
-                        response.headers.get("Content-Disposition", "")
+                    filepath = posixpath.join(
+                        parent_dir, _get_filename_from_response(response, url)
                     )
-                    filepath = posixpath.join(parent_dir, params["filename"])
                 else:
                     raise Exception(
                         f"Downloading {url} failed with status {response.status}:\n{response.content}"
@@ -125,9 +132,8 @@ async def download(
                     connector.run(
                         location=location,
                         command=[
-                            'if [ command -v curl ]; curl -L -o "{path}"; else wget -P "{dir}" {url}; fi'.format(
-                                dir=parent_dir, path=filepath, url=url
-                            )
+                            f'if [ command -v curl ]; then curl -L -o "{filepath}" "{url}"; '
+                            f'else wget -P "{parent_dir}" "{url}"; fi'
                         ],
                     )
                 )

streamflow/deployment/connector/container.py

@@ -123,7 +123,7 @@ def __init__(
             cacheSize = resourcesCacheSize
             if cacheSize is not None:
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         "The `resourcesCacheSize` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                         "Use `locationsCacheSize` instead."
                     )
@@ -134,7 +134,7 @@ def __init__(
             cacheTTL = resourcesCacheTTL
             if cacheTTL is not None:
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         "The `resourcesCacheTTL` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                         "Use `locationsCacheTTL` instead."
                     )

streamflow/deployment/connector/kubernetes.py

@@ -189,7 +189,7 @@ def __init__(
             cacheSize = resourcesCacheSize
             if cacheSize is not None:
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         "The `resourcesCacheSize` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                         "Use `locationsCacheSize` instead."
                     )
@@ -200,7 +200,7 @@ def __init__(
             cacheTTL = resourcesCacheTTL
             if cacheTTL is not None:
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         "The `resourcesCacheTTL` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                         "Use `locationsCacheTTL` instead."
                     )

streamflow/deployment/connector/queue_manager.py

@@ -341,7 +341,7 @@ def __init__(
         self._inner_ssh_connector: bool = False
         if hostname is not None:
             if logger.isEnabledFor(logging.WARN):
-                logger.warn(
+                logger.warning(
                     "Inline SSH options are deprecated and will be removed in StreamFlow 0.3.0. "
                     f"Define a standalone `SSHConnector` and link the `{self.__class__.__name__}` "
                     "to it using the `wraps` property."
@@ -375,7 +375,7 @@ def __init__(
                         files_map[name] = f.read()
         if file is not None:
             if logger.isEnabledFor(logging.WARN):
-                logger.warn(
+                logger.warning(
                     "The `file` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                     "Use `services` instead."
                 )
@@ -553,12 +553,12 @@ async def undeploy(self, external: bool) -> None:
         self.scheduledJobs = {}
         if self._inner_ssh_connector:
             if logger.isEnabledFor(logging.INFO):
-                logger.warn(
+                logger.warning(
                     f"UNDEPLOYING inner SSH connector for {self.deployment_name} deployment."
                 )
             await self.connector.undeploy(external)
             if logger.isEnabledFor(logging.INFO):
-                logger.warn(
+                logger.warning(
                     f"COMPLETED Undeployment of inner SSH connector for {self.deployment_name} deployment."
                 )
 

streamflow/deployment/connector/ssh.py

@@ -284,7 +284,7 @@ def __init__(
                     services_map[name] = f.read()
         if file is not None:
             if logger.isEnabledFor(logging.WARN):
-                logger.warn(
+                logger.warning(
                     "The `file` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                     "Use `services` instead."
                 )

streamflow/deployment/manager.py

@@ -147,7 +147,7 @@ async def _inner_deploy(
         else:
             if deployment_config.wraps is not None:
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         f"The `wraps` directive has no effect on deployment {deployment_config.name}, "
                         f"as the `{deployment_config.type}` connector does not inherit from the ConnectorWrapper class."
                     )

streamflow/deployment/utils.py

@@ -31,7 +31,7 @@ def get_binding_config(
             else:
                 target_deployment = workflow_config.deplyoments[target["model"]]
                 if logger.isEnabledFor(logging.WARN):
-                    logger.warn(
+                    logger.warning(
                         "The `model` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                         "Use `deployment` instead."
                     )
@@ -40,7 +40,7 @@ def get_binding_config(
                 locations = target.get("resources")
                 if locations is not None:
                     if logger.isEnabledFor(logging.WARN):
-                        logger.warn(
+                        logger.warning(
                             "The `resources` keyword is deprecated and will be removed in StreamFlow 0.3.0. "
                             "Use `locations` instead."
                         )

streamflow/ext/plugin.py

@@ -49,7 +49,7 @@ def _register(self, name: str, cls: type, extension_point: str):
         )
         if name in extension_points[extension_point]:
             if logger.isEnabledFor(logging.WARN):
-                logger.warn(
+                logger.warning(
                     "{} is already installed and will be overridden by {}".format(
                         name, self.__class__.__module__ + "." + self.__class__.__name__
                     )

streamflow/provenance/run_crate.py

@@ -571,7 +571,7 @@ async def add_file(self, file: MutableMapping[str, str]) -> None:
         dst_parent = posixpath.dirname(posixpath.normpath(dst))
         if src in self.files_map:
             if logger.isEnabledFor(logging.WARN):
-                logger.warn(f"File {src} is already present in the archive.")
+                logger.warning(f"File {src} is already present in the archive.")
         else:
             if os.path.isfile(os.path.realpath(src)):
                 checksum = _file_checksum(
@@ -681,7 +681,7 @@ async def add_property(self, key: str, value: str):
             current_obj = current_obj[k]
         if logger.isEnabledFor(logging.WARN):
             if keys[-1] in current_obj:
-                logger.warn(
+                logger.warning(
                     f"Key {key} already exists in archive manifest and will be overridden."
                 )
         value = ESCAPED_EQUAL.sub("=", value)
@@ -889,7 +889,7 @@ async def create_archive(
                     if dst not in archive.namelist():
                         archive.write(src, dst)
                 else:
-                    logger.warn(f"File {src} does not exist.")
+                    logger.warning(f"File {src} does not exist.")
         print(f"Successfully created run_crate archive at {path}")
 
     @abstractmethod