Remove conditional logic and checks for Python 3.7 (#13285)

src/integrations/prefect-aws/tests/deployments/test_steps.py

@@ -186,7 +186,6 @@ def test_push_pull_empty_folders(s3_setup, tmp_path, mock_aws_credentials):
     assert not (tmp_path / "empty2_copy").exists()
 
 
-@pytest.mark.skipif(sys.version_info < (3, 8), reason="requires Python 3.8+")
 def test_s3_session_with_params():
     with patch("boto3.Session") as mock_session:
         get_s3_client(

src/integrations/prefect-dask/tests/test_task_runners.py

@@ -33,19 +33,11 @@ def event_loop(request):
     see https://github.com/pytest-dev/pytest-asyncio/issues/68
     When running on Windows we need to use a non-default loop for subprocess support.
     """
-    if sys.platform == "win32" and sys.version_info >= (3, 8):
+    if sys.platform == "win32":
         asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
 
     policy = asyncio.get_event_loop_policy()
 
-    if sys.version_info < (3, 8) and sys.platform != "win32":
-        from prefect.utilities.compat import ThreadedChildWatcher
-
-        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can
-        # lead to errors in tests as the previous default `SafeChildWatcher`  is not
-        # compatible with threaded event loops.
-        policy.set_child_watcher(ThreadedChildWatcher())
-
     loop = policy.new_event_loop()
 
     # configure asyncio logging to capture long running tasks

src/integrations/prefect-docker/tests/conftest.py

@@ -50,19 +50,11 @@ def event_loop(request):
 
     When running on Windows we need to use a non-default loop for subprocess support.
     """
-    if sys.platform == "win32" and sys.version_info >= (3, 8):
+    if sys.platform == "win32":
         asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
 
     policy = asyncio.get_event_loop_policy()
 
-    if sys.version_info < (3, 8) and sys.platform != "win32":
-        from prefect.utilities.compat import ThreadedChildWatcher
-
-        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can
-        # lead to errors in tests as the previous default `SafeChildWatcher`  is not
-        # compatible with threaded event loops.
-        policy.set_child_watcher(ThreadedChildWatcher())
-
     loop = policy.new_event_loop()
 
     # configure asyncio logging to capture long running tasks

src/integrations/prefect-docker/tests/deployments/test_steps.py

@@ -233,10 +233,7 @@ def test_real_auto_dockerfile_build(docker_client_with_cleanup):
             labels=["prefect-docker-test"],
             remove=True,
         )
-        if sys.version_info >= (3, 8):
-            assert "2" in output.decode()
-        else:
-            assert "1" in output.decode()
+        assert "2" in output.decode()
 
     finally:
         docker_client_with_cleanup.containers.prune(

src/integrations/prefect-ray/tests/test_task_runners.py

@@ -35,7 +35,7 @@ def event_loop(request):
     see https://github.com/pytest-dev/pytest-asyncio/issues/68
     When running on Windows we need to use a non-default loop for subprocess support.
     """
-    if sys.platform == "win32" and sys.version_info >= (3, 8):
+    if sys.platform == "win32":
         asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
 
     policy = asyncio.get_event_loop_policy()

src/prefect/__init__.py

@@ -25,18 +25,6 @@
 
 del _version, pathlib
 
-if sys.version_info < (3, 8):
-    warnings.warn(
-        (
-            "Prefect dropped support for Python 3.7 when it reached end-of-life"
-            " . To use new versions of Prefect, you will need"
-            " to upgrade to Python 3.8+. See https://devguide.python.org/versions/ for "
-            " more details."
-        ),
-        FutureWarning,
-        stacklevel=2,
-    )
-
 
 # Import user-facing API
 from prefect.runner import Runner, serve

src/prefect/infrastructure/process.py

@@ -7,7 +7,6 @@
 For upgrade instructions, see https://docs.prefect.io/latest/guides/upgrade-guide-agents-to-workers/.
 """
 
-import asyncio
 import contextlib
 import os
 import shlex
@@ -21,7 +20,6 @@
 
 import anyio
 import anyio.abc
-import sniffio
 
 from prefect._internal.compatibility.deprecated import deprecated_class
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
@@ -43,20 +41,6 @@
     STATUS_CONTROL_C_EXIT = 0xC000013A
 
 
-def _use_threaded_child_watcher():
-    if (
-        sys.version_info < (3, 8)
-        and sniffio.current_async_library() == "asyncio"
-        and sys.platform != "win32"
-    ):
-        from prefect.utilities.compat import ThreadedChildWatcher
-
-        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can
-        # lead to errors in tests on unix as the previous default `SafeChildWatcher`
-        # is not compatible with threaded event loops.
-        asyncio.get_event_loop_policy().set_child_watcher(ThreadedChildWatcher())
-
-
 def _infrastructure_pid_from_process(process: anyio.abc.Process) -> str:
     hostname = socket.gethostname()
     return f"{hostname}:{process.pid}"
@@ -121,7 +105,6 @@ async def run(
         if not self.command:
             raise ValueError("Process cannot be run with empty command.")
 
-        _use_threaded_child_watcher()
         display_name = f" {self.name!r}" if self.name else ""
 
         # Open a subprocess to execute the flow run

src/prefect/logging/formatters.py

@@ -99,10 +99,7 @@ def __init__(
             style_kwargs["defaults"] = defaults
 
         # validate added in 3.8
-        if sys.version_info >= (3, 8):
-            init_kwargs["validate"] = validate
-        else:
-            validate = False
+        init_kwargs["validate"] = validate
 
         super().__init__(format, datefmt, style, **init_kwargs)
 

src/prefect/runner/runner.py

@@ -51,7 +51,6 @@ def fast_flow():
 import anyio
 import anyio.abc
 import pendulum
-import sniffio
 from rich.console import Console, Group
 from rich.table import Table
 
@@ -550,7 +549,6 @@ async def _run_process(
         if sys.platform == "win32":
             kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
 
-        _use_threaded_child_watcher()
         flow_run_logger.info("Opening process...")
 
         env = get_current_settings().to_environment_variables(exclude_unset=True)
@@ -1190,20 +1188,6 @@ def __repr__(self):
     STATUS_CONTROL_C_EXIT = 0xC000013A
 
 
-def _use_threaded_child_watcher():
-    if (
-        sys.version_info < (3, 8)
-        and sniffio.current_async_library() == "asyncio"
-        and sys.platform != "win32"
-    ):
-        from prefect.utilities.compat import ThreadedChildWatcher
-
-        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can
-        # lead to errors in tests on unix as the previous default `SafeChildWatcher`
-        # is not compatible with threaded event loops.
-        asyncio.get_event_loop_policy().set_child_watcher(ThreadedChildWatcher())
-
-
 @sync_compatible
 async def serve(
     *args: RunnerDeployment,

src/prefect/testing/utilities.py

@@ -2,7 +2,6 @@
 Internal utilities for tests.
 """
 
-import sys
 import warnings
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
@@ -35,52 +34,10 @@ def exceptions_equal(a, b):
 
 
 # AsyncMock has a new import path in Python 3.8+
-
-if sys.version_info < (3, 8):
-    # https://docs.python.org/3/library/unittest.mock.html#unittest.mock.AsyncMock
-    from mock import AsyncMock  # noqa
-else:
-    from unittest.mock import AsyncMock  # noqa
+from unittest.mock import AsyncMock  # noqa
 
 # MagicMock supports async magic methods in Python 3.8+
-
-if sys.version_info < (3, 8):
-    from unittest.mock import MagicMock as _MagicMock
-    from unittest.mock import MagicProxy as _MagicProxy
-
-    class MagicMock(_MagicMock):
-        def _mock_set_magics(self):
-            """Patch to include proxies for async methods"""
-            super()._mock_set_magics()
-
-            for attr in {"__aenter__", "__aexit__", "__anext__"}:
-                if not hasattr(MagicMock, attr):
-                    setattr(MagicMock, attr, _MagicProxy(attr, self))
-
-        def _get_child_mock(self, **kw):
-            """Patch to return async mocks for async methods"""
-            # This implementation copied from unittest in Python 3.8
-            _new_name = kw.get("_new_name")
-            if _new_name in self.__dict__.get("_spec_asyncs", {}):
-                return AsyncMock(**kw)
-
-            _type = type(self)
-            if issubclass(_type, MagicMock) and _new_name in {
-                "__aenter__",
-                "__aexit__",
-                "__anext__",
-            }:
-                if self._mock_sealed:
-                    attribute = "." + kw["name"] if "name" in kw else "()"
-                    mock_name = self._extract_mock_name() + attribute
-                    raise AttributeError(mock_name)
-
-                return AsyncMock(**kw)
-
-            return super()._get_child_mock(**kw)
-
-else:
-    from unittest.mock import MagicMock
+from unittest.mock import MagicMock  # noqa
 
 
 def kubernetes_environments_equal(

src/prefect/utilities/annotations.py

@@ -1,4 +1,3 @@
-import sys
 import warnings
 from abc import ABC
 from collections import namedtuple
@@ -17,13 +16,7 @@ class BaseAnnotation(
     """
 
     def unwrap(self) -> T:
-        if sys.version_info < (3, 8):
-            # cannot simply return self.value due to recursion error in Python 3.7
-            # also _asdict does not follow convention; it's not an internal method
-            # https://stackoverflow.com/a/26180604
-            return self._asdict()["value"]
-        else:
-            return self.value
+        return self.value
 
     def rewrap(self, value: T) -> "BaseAnnotation[T]":
         return type(self)(value)

src/prefect/workers/process.py

@@ -14,7 +14,6 @@
 checkout out the [Prefect docs](/concepts/work-pools/).
 """
 
-import asyncio
 import contextlib
 import os
 import signal
@@ -27,7 +26,6 @@
 
 import anyio
 import anyio.abc
-import sniffio
 
 from prefect._internal.pydantic import HAS_PYDANTIC_V2
 
@@ -56,20 +54,6 @@
     STATUS_CONTROL_C_EXIT = 0xC000013A
 
 
-def _use_threaded_child_watcher():
-    if (
-        sys.version_info < (3, 8)
-        and sniffio.current_async_library() == "asyncio"
-        and sys.platform != "win32"
-    ):
-        from prefect.utilities.compat import ThreadedChildWatcher
-
-        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can
-        # lead to errors in tests on unix as the previous default `SafeChildWatcher`
-        # is not compatible with threaded event loops.
-        asyncio.get_event_loop_policy().set_child_watcher(ThreadedChildWatcher())
-
-
 def _infrastructure_pid_from_process(process: anyio.abc.Process) -> str:
     hostname = socket.gethostname()
     return f"{hostname}:{process.pid}"
@@ -168,7 +152,6 @@ async def run(
         if sys.platform == "win32":
             kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
 
-        _use_threaded_child_watcher()
         flow_run_logger.info("Opening process...")
 
         working_dir_ctx = (

tests/cli/test_deploy.py

@@ -86,52 +86,30 @@ def readchar():
 @pytest.fixture
 def project_dir(tmp_path):
     with tmpchdir(tmp_path):
-        if sys.version_info >= (3, 8):
-            shutil.copytree(TEST_PROJECTS_DIR, tmp_path, dirs_exist_ok=True)
-            prefect_home = tmp_path / ".prefect"
-            prefect_home.mkdir(exist_ok=True, mode=0o0700)
-            initialize_project()
-            yield tmp_path
-        else:
-            shutil.copytree(TEST_PROJECTS_DIR, tmp_path / "three-seven")
-            prefect_home = tmp_path / "three-seven" / ".prefect"
-            prefect_home.mkdir(exist_ok=True, mode=0o0700)
-            initialize_project()
-            yield tmp_path / "three-seven"
+        shutil.copytree(TEST_PROJECTS_DIR, tmp_path, dirs_exist_ok=True)
+        prefect_home = tmp_path / ".prefect"
+        prefect_home.mkdir(exist_ok=True, mode=0o0700)
+        initialize_project()
+        yield tmp_path
 
 
 @pytest.fixture
 def project_dir_with_single_deployment_format(tmp_path):
     with tmpchdir(tmp_path):
-        if sys.version_info >= (3, 8):
-            shutil.copytree(TEST_PROJECTS_DIR, tmp_path, dirs_exist_ok=True)
-            prefect_home = tmp_path / ".prefect"
-            prefect_home.mkdir(exist_ok=True, mode=0o0700)
-            initialize_project()
-
-            with open("prefect.yaml", "r") as f:
-                contents = yaml.safe_load(f)
-
-            contents["deployments"][0]["schedule"] = None
-
-            with open("deployment.yaml", "w") as f:
-                yaml.safe_dump(contents["deployments"][0], f)
+        shutil.copytree(TEST_PROJECTS_DIR, tmp_path, dirs_exist_ok=True)
+        prefect_home = tmp_path / ".prefect"
+        prefect_home.mkdir(exist_ok=True, mode=0o0700)
+        initialize_project()
 
-            yield tmp_path
-        else:
-            shutil.copytree(TEST_PROJECTS_DIR, tmp_path / "three-seven")
-            (tmp_path / "three-seven" / ".prefect").mkdir(exist_ok=True, mode=0o0700)
-            initialize_project()
-
-            with open("prefect.yaml", "r") as f:
-                contents = yaml.safe_load(f)
+        with open("prefect.yaml", "r") as f:
+            contents = yaml.safe_load(f)
 
-            contents["deployments"][0]["schedule"] = None
+        contents["deployments"][0]["schedule"] = None
 
-            with open("deployment.yaml", "w") as f:
-                yaml.safe_dump(contents["deployments"][0], f)
+        with open("deployment.yaml", "w") as f:
+            yaml.safe_dump(contents["deployments"][0], f)
 
-            yield tmp_path / "three-seven"
+        yield tmp_path
 
 
 @pytest.fixture