Merge branch 'v4.4.0-rc-dev'

Author: BryanFauble

.github/workflows/build.yml

@@ -398,7 +398,7 @@ jobs:
 
     strategy:
       matrix:
-        os: [ubuntu-20.04, macos-11, windows-2019]
+        os: [ubuntu-20.04, macos-12, windows-2022]
 
         # python versions should be consistent with the strategy matrix and the runs-integration-tests versions
         python: [3.8, '3.9', '3.10', '3.11']

Pipfile.lock

@@ -35,6 +35,7 @@ stored in the `~/.synapseConfig` file.
 -  \[[SYNPY-1483](https://sagebionetworks.jira.com/browse/SYNPY-1483)\] - Update credential chain to use user args first
 -  \[[SYNPY-1485](https://sagebionetworks.jira.com/browse/SYNPY-1485)\] - Include isort with pre-commit
 -  \[[SYNPY-1487](https://sagebionetworks.jira.com/browse/SYNPY-1487)\] - Fix missing coverage.xml bug
+-  \[[SYNPY-1504](https://sagebionetworks.jira.com/browse/SYNPY-1504)\] - Review usage of download_location and path parameters on File class
 
 ## 4.3.0 (2024-05-30)
 ### Highlights

docs/news.md

@@ -131,7 +131,7 @@ def store_file():
     # 5. Downloading a file ===============================================================
     # Downloading a file to a location has a default beahvior of "keep.both"
     downloaded_file = File(
-        id=file.id, download_location=os.path.expanduser("~/temp/myNewFolder")
+        id=file.id, path=os.path.expanduser("~/temp/myNewFolder")
     ).get()
     print(f"Downloaded file: {downloaded_file.path}")
 
@@ -142,7 +142,7 @@ def store_file():
     print(f"Before file md5: {utils.md5_for_file(path_to_file).hexdigest()}")
     downloaded_file = File(
         id=downloaded_file.id,
-        download_location=os.path.expanduser("~/temp/myNewFolder"),
+        path=os.path.expanduser("~/temp/myNewFolder"),
         if_collision="overwrite.local",
     ).get()
     print(f"After file md5: {utils.md5_for_file(path_to_file).hexdigest()}")
@@ -153,7 +153,7 @@ def store_file():
     print(f"Before file md5: {utils.md5_for_file(path_to_file).hexdigest()}")
     downloaded_file = File(
         id=downloaded_file.id,
-        download_location=os.path.expanduser("~/temp/myNewFolder"),
+        path=os.path.expanduser("~/temp/myNewFolder"),
         if_collision="keep.local",
     ).get()
     print(f"After file md5: {utils.md5_for_file(path_to_file).hexdigest()}")

docs/scripts/object_orientated_programming_poc/oop_poc_file.py

@@ -56,7 +56,6 @@ install_requires =
     asyncio-atexit~=1.0.1
     httpx~=0.27.0
     tqdm>=4.66.2,<5.0
-    loky~=3.0.0
     async-lru~=2.0.4
     psutil~=5.9.8
 tests_require =

setup.cfg

@@ -34,7 +34,6 @@
 import httpx
 import requests
 from deprecated import deprecated
-from loky import get_reusable_executor
 from opentelemetry import trace
 from opentelemetry.trace import SpanKind
 
@@ -410,7 +409,6 @@ def log_response(response: httpx.Response) -> None:
         self._thread_executor = {}
         self._process_executor = {}
         self._parallel_file_transfer_semaphore = {}
-        self._md5_semaphore = {}
         self.use_boto_sts_transfers = transfer_config["use_boto_sts"]
         self._parts_transfered_counter = 0
 
@@ -522,55 +520,6 @@ def close_pool() -> None:
         asyncio_atexit.register(close_pool)
         return self._thread_executor[asyncio_event_loop]
 
-    def _get_process_pool_executor(self, asyncio_event_loop: asyncio.AbstractEventLoop):
-        """
-        Retrieve the process pool executor for the Synapse client. Or create a new one
-        if it does not exist. This executor is used for parallel processing of data.
-
-        This is expected to be called from within an AsyncIO loop.
-
-        Note: Within Windows a ProcessPoolExecutor requires that the initial entry point
-        into the code be within a `if __name__ == "__main__":` block. This is not
-        possible within the current codebase as it would require everyone using this
-        library to have this as their entry point. As a result, the ProcessPoolExecutor
-        will not work within Windows.
-
-        To get around this Windows limitation this is using this package:
-        https://github.com/joblib/loky
-        """
-        if (
-            hasattr(self, "_process_executor")
-            and asyncio_event_loop in self._process_executor
-            and self._process_executor[asyncio_event_loop] is not None
-        ):
-            return self._process_executor[asyncio_event_loop]
-
-        self._process_executor.update({asyncio_event_loop: get_reusable_executor(1)})
-
-        return self._process_executor[asyncio_event_loop]
-
-    def _get_md5_semaphore(
-        self, asyncio_event_loop: asyncio.AbstractEventLoop
-    ) -> asyncio.Semaphore:
-        """
-        Retrieve the semaphore for the Synapse client. Or create a new one if it does not
-        exist. This semaphore is used to ensure that only one process is calculating the
-        MD5 hash at a time. This is to prevent the custom process pool executor from
-        thrashing or handling the waiting. We should let asyncio handle the waiting.
-
-        This is expected to be called from within an AsyncIO loop.
-        """
-        if (
-            hasattr(self, "_md5_semaphore")
-            and asyncio_event_loop in self._md5_semaphore
-            and self._md5_semaphore[asyncio_event_loop] is not None
-        ):
-            return self._md5_semaphore[asyncio_event_loop]
-
-        self._md5_semaphore.update({asyncio_event_loop: asyncio.Semaphore(1)})
-
-        return self._md5_semaphore[asyncio_event_loop]
-
     def _get_parallel_file_transfer_semaphore(
         self, asyncio_event_loop: asyncio.AbstractEventLoop
     ) -> asyncio.Semaphore:

synapseclient/client.py

@@ -473,6 +473,7 @@ def _stream_and_write_chunk(
             retry_errors=RETRYABLE_CONNECTION_ERRORS,
             retry_exceptions=RETRYABLE_CONNECTION_EXCEPTIONS,
             retry_max_back_off=DEFAULT_MAX_BACK_OFF_ASYNC,
+            read_response_content=False,
         )
 
         return start, end
@@ -531,7 +532,9 @@ def _execute_stream_and_write_chunk(
     with session.stream(
         method="GET", url=presigned_url_provider.get_info().url, headers=range_header
     ) as response:
-        _raise_for_status_httpx(response=response, logger=request._syn.logger)
+        _raise_for_status_httpx(
+            response=response, logger=request._syn.logger, read_response_content=False
+        )
         data = response.read()
         data_length = len(data)
         request._write_chunk(

synapseclient/core/download/download_async.py

@@ -191,7 +191,10 @@ def _raise_for_status(response, verbose=False):
 
 
 def _raise_for_status_httpx(
-    response: httpx.Response, logger: logging.Logger, verbose: bool = False
+    response: httpx.Response,
+    logger: logging.Logger,
+    verbose: bool = False,
+    read_response_content: bool = True,
 ) -> None:
     """
     Replacement for requests.response.raise_for_status().
@@ -200,8 +203,12 @@ def _raise_for_status_httpx(
 
     Arguments:
         response: The response object from the HTTPX request.
+        logger: The logger object to log any exceptions.
         verbose: If True, the request and response information will be appended to the
             error message.
+        read_response_content: If True, the response content will be read and appended
+            to the error message. If False, the response content will not be read and
+            appended to the error message.
     """
 
     message = None
@@ -242,7 +249,7 @@ def _raise_for_status_httpx(
         # 450: 'blocked_by_windows_parental_controls'
         # 451: 'unavailable_for_legal_reasons'
         # 499: 'client_closed_request'
-        message_body = _get_message(response, logger)
+        message_body = _get_message(response, logger) if read_response_content else ""
         message = f"{response.status_code} {CLIENT_ERROR} {message_body}"
 
     elif 500 <= response.status_code < 600:
@@ -257,7 +264,7 @@ def _raise_for_status_httpx(
         # 507: 'insufficient_storage'
         # 509: 'bandwidth_limit_exceeded'
         # 510: 'not_extended'
-        message_body = _get_message(response, logger)
+        message_body = _get_message(response, logger) if read_response_content else ""
         message = f"{response.status_code} {SERVER_ERROR} {message_body}"
 
     if message is not None:
@@ -275,7 +282,8 @@ def _raise_for_status_httpx(
                 # Append the response received
                 message += f"\n\n{RESPONSE_PREFIX}\n{str(response)}"
                 message += f"\n{HEADERS_PREFIX}{response.headers}"
-                message += f"\n{BODY_PREFIX}{message_body}\n\n"
+                if read_response_content:
+                    message += f"\n{BODY_PREFIX}{message_body}\n\n"
             except Exception:  # noqa
                 logger.exception(UNABLE_TO_APPEND_RESPONSE)
                 message += f"\n{UNABLE_TO_APPEND_RESPONSE}"

synapseclient/core/exceptions.py

@@ -64,6 +64,11 @@
     "timeout",
     "ReadError",
     "ReadTimeout",
+    # HTTPX Specific connection exceptions:
+    "RemoteProtocolError",
+    "TimeoutException",
+    "ConnectError",
+    "ConnectTimeout",
 ]
 
 DEBUG_EXCEPTION = "calling %s resulted in an Exception"
@@ -272,6 +277,7 @@ async def with_retry_time_based_async(
     retry_back_off_factor: float = DEFAULT_BACK_OFF_FACTOR_ASYNC,
     retry_max_back_off: float = DEFAULT_MAX_BACK_OFF_ASYNC,
     retry_max_wait_before_failure: float = DEFAULT_MAX_WAIT_BEFORE_FAIL_ASYNC,
+    read_response_content: bool = True,
 ) -> Union[Exception, httpx.Response, Any, None]:
     """
     Retries the given function under certain conditions. This is created such that it
@@ -297,6 +303,7 @@ async def with_retry_time_based_async(
         retry_back_off_factor: The factor to increase the wait time by for each retry.
         retry_max_back_off: The maximum wait time.
         retry_max_wait_before_failure: The maximum wait time before failure.
+        read_response_content: Whether to read the response content for HTTP requests.
 
     Example: Using with_retry
         Using ``with_retry_time_based_async`` to consolidate inputs into a list.
@@ -353,7 +360,10 @@ async def foo(a, b, c): return [a, b, c]
         retries += 1
         if total_wait < retry_max_wait_before_failure and retry:
             _log_for_retry(
-                logger=logger, response=response, caught_exception=caught_exception
+                logger=logger,
+                response=response,
+                caught_exception=caught_exception,
+                read_response_content=read_response_content,
             )
 
             backoff_wait = calculate_exponential_backoff(
@@ -395,6 +405,7 @@ def with_retry_time_based(
     retry_back_off_factor: float = DEFAULT_BACK_OFF_FACTOR_ASYNC,
     retry_max_back_off: float = DEFAULT_MAX_BACK_OFF_ASYNC,
     retry_max_wait_before_failure: float = DEFAULT_MAX_WAIT_BEFORE_FAIL_ASYNC,
+    read_response_content: bool = True,
 ) -> Union[Exception, httpx.Response, Any, None]:
     """
     Retries the given function under certain conditions. This is created such that it
@@ -420,6 +431,7 @@ def with_retry_time_based(
         retry_back_off_factor: The factor to increase the wait time by for each retry.
         retry_max_back_off: The maximum wait time.
         retry_max_wait_before_failure: The maximum wait time before failure.
+        read_response_content: Whether to read the response content for HTTP requests.
 
     Example: Using with_retry
         Using ``with_retry_time_based`` to consolidate inputs into a list.
@@ -476,7 +488,10 @@ async def foo(a, b, c): return [a, b, c]
         retries += 1
         if total_wait < retry_max_wait_before_failure and retry:
             _log_for_retry(
-                logger=logger, response=response, caught_exception=caught_exception
+                logger=logger,
+                response=response,
+                caught_exception=caught_exception,
+                read_response_content=read_response_content,
             )
 
             backoff_wait = calculate_exponential_backoff(
@@ -569,16 +584,18 @@ def _log_for_retry(
     logger: logging.Logger,
     response: httpx.Response = None,
     caught_exception: Exception = None,
+    read_response_content: bool = True,
 ) -> None:
     """Logs the retry message to debug.
 
     Arguments:
         logger: The logger to use for logging the retry message.
         response: The response object from the request.
         caught_exception: The exception caught from the request.
+        read_response_content: Whether to read the response content for HTTP requests.
     """
     if response is not None:
-        response_message = _get_message(response)
+        response_message = _get_message(response) if read_response_content else ""
         url_message_part = ""
 
         if hasattr(response, "request") and hasattr(response.request, "url"):

synapseclient/core/retry.py

@@ -113,7 +113,7 @@
 )
 from synapseclient.core.utils import MB
 from synapseclient.core.utils import md5_fn as md5_fn_util
-from synapseclient.core.utils import md5_for_file_multiprocessing
+from synapseclient.core.utils import md5_for_file_hex
 
 if TYPE_CHECKING:
     from synapseclient import Synapse
@@ -668,17 +668,7 @@ async def multipart_upload_file_async(
         mime_type, _ = mimetypes.guess_type(file_path, strict=False)
         content_type = mime_type or "application/octet-stream"
 
-    md5_hex = md5 or (
-        await md5_for_file_multiprocessing(
-            filename=file_path,
-            process_pool_executor=syn._get_process_pool_executor(
-                asyncio_event_loop=asyncio.get_running_loop()
-            ),
-            md5_semaphore=syn._get_md5_semaphore(
-                asyncio_event_loop=asyncio.get_running_loop()
-            ),
-        )
-    )
+    md5_hex = md5 or md5_for_file_hex(filename=file_path)
 
     part_size = get_part_size(
         part_size or DEFAULT_PART_SIZE,