def _get_max_in_flight_requests()

in src/azstoragetorch/_client.py [0:0]

• 3 lines of code
• 6 McCabe index (conditional complexity)

    def _get_max_in_flight_requests(self) -> int:
        # Ideally we would just match this value to the max workers of the executor. However
        # the executor class does not publicly expose its max worker count. So, instead we copy
        # the max worker calculation from the executor class and inject it into both the executor
        # and semaphore
        #
        # In Python 3.13, os.process_cpu_count() was added and the ThreadPoolExecutor updated to
        # use os.process_cpu_count() instead of os.cpu_count() when calculating default max workers.
        # To match ThreadPoolExecutor defaults across Python versions, we use process_cpu_count
        # if available, otherwise fall back to os.cpu_count().
        cpu_count_fn = getattr(os, "process_cpu_count", os.cpu_count)
        return min(32, (cpu_count_fn() or 1) + 4)