PYTHON-6158 Clean up version-gated code paths

pymongo/asynchronous/helpers.py

@@ -20,7 +20,6 @@
 import builtins
 import functools
 import socket
-import sys
 from typing import (
     Any,
     Callable,
@@ -94,15 +93,5 @@ async def _getaddrinfo(
         return socket.getaddrinfo(host, port, **kwargs)
 
 
-if sys.version_info >= (3, 10):
-    anext = builtins.anext
-    aiter = builtins.aiter
-else:
-
-    async def anext(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#anext."""
-        return await cls.__anext__()
-
-    def aiter(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#anext."""
-        return cls.__aiter__()
+anext = builtins.anext
+aiter = builtins.aiter

pymongo/cursor_shared.py

@@ -43,19 +43,14 @@
 from pymongo.write_concern import validate_boolean
 
 if TYPE_CHECKING:
-    import sys
+    from typing import Self
 
     from _typeshed import SupportsItems
 
     from bson.codec_options import CodecOptions
     from pymongo.message import _OpMsg
     from pymongo.read_preferences import _ServerMode
 
-    if sys.version_info >= (3, 11):
-        from typing import Self
-    else:
-        from typing_extensions import Self
-
 _CURSOR_DOC_FIELDS = {"cursor": {"firstBatch": 1, "nextBatch": 1}}
 
 

pymongo/network_layer.py

@@ -33,7 +33,6 @@
 )
 
 from pymongo import _csot, ssl_support
-from pymongo._asyncio_task import create_task
 from pymongo.common import MAX_MESSAGE_SIZE
 from pymongo.compression_support import decompress
 from pymongo.errors import ProtocolError, _OperationCancelled
@@ -711,8 +710,8 @@ async def async_receive_message(
         # timeouts on AWS Lambda and other FaaS environments.
         timeout = max(deadline - time.monotonic(), 0)
 
-    cancellation_task = create_task(_poll_cancellation(conn))
-    read_task = create_task(conn.conn.get_conn.read(request_id, max_message_size))
+    cancellation_task = asyncio.create_task(_poll_cancellation(conn))
+    read_task = asyncio.create_task(conn.conn.get_conn.read(request_id, max_message_size))
     tasks = [read_task, cancellation_task]
     try:
         done, pending = await asyncio.wait(

pymongo/periodic_executor.py

@@ -24,7 +24,6 @@
 from typing import Any, Optional
 
 from pymongo import _csot, _op_id
-from pymongo._asyncio_task import create_task
 from pymongo.lock import _create_lock
 
 _IS_SYNC = False
@@ -65,9 +64,9 @@ def open(self) -> None:
         self._stopped = False
 
         if self._task is None or (
-            self._task.done() and not self._task.cancelled() and not self._task.cancelling()  # type: ignore[unused-ignore, attr-defined]
+            self._task.done() and not self._task.cancelled() and not self._task.cancelling()
         ):
-            self._task = create_task(self._run(), name=self._name)
+            self._task = asyncio.create_task(self._run(), name=self._name)
 
     def close(self, dummy: Any = None) -> None:
         """Stop. To restart, call open().
@@ -98,7 +97,7 @@ async def _run(self) -> None:
         _csot.reset_all()
         _op_id.reset()
         while not self._stopped:
-            if self._task and self._task.cancelling():  # type: ignore[unused-ignore, attr-defined]
+            if self._task and self._task.cancelling():
                 raise asyncio.CancelledError
             try:
                 if not await self._target():

pymongo/pool_options.py

@@ -472,10 +472,7 @@ def _update_metadata(self, driver: DriverInfo) -> None:
             # raises.
             names = metadata["driver"]["name"].split("|")
             versions = metadata["driver"]["version"].split("|")
-            if sys.version_info >= (3, 10):
-                pairs = zip(names, versions, strict=True)
-            else:
-                pairs = zip(names, versions)
+            pairs = zip(names, versions, strict=True)
             if (driver.name, driver.version) in pairs:
                 self.__appended_drivers.add(driver)
 

pymongo/pool_shared.py

@@ -417,11 +417,11 @@ async def _configured_protocol_interface(
 
     host = address[0]
     # asyncio does not support TLS session resumption natively (cpython#79152,
-    # closed without a fix).  On Python 3.11+ SSLProtocol.__init__ calls
-    # wrap_bio() synchronously before the first event-loop yield, so setting
-    # sslobject_class is race-free.  Session injection is skipped on older
-    # Python versions.  (The async path always uses stdlib ssl, never PyOpenSSL.)
-    if ssl_session_cache is not None and sys.version_info >= (3, 11):
+    # closed without a fix).  SSLProtocol.__init__ calls wrap_bio()
+    # synchronously before the first event-loop yield, so setting
+    # sslobject_class is race-free.  (The async path always uses stdlib ssl,
+    # never PyOpenSSL.)
+    if ssl_session_cache is not None:
         session = ssl_session_cache[0]
         if session is not None:
             _session = session

pymongo/synchronous/helpers.py

@@ -20,7 +20,6 @@
 import builtins
 import functools
 import socket
-import sys
 from typing import (
     Any,
     Callable,
@@ -94,15 +93,5 @@ def _getaddrinfo(
         return socket.getaddrinfo(host, port, **kwargs)
 
 
-if sys.version_info >= (3, 10):
-    next = builtins.next
-    iter = builtins.iter
-else:
-
-    def next(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#next."""
-        return cls.__next__()
-
-    def iter(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#next."""
-        return cls.__iter__()
+next = builtins.next
+iter = builtins.iter

test/asynchronous/helpers.py

@@ -24,7 +24,6 @@
 
 from bson import SON
 from pymongo import common
-from pymongo._asyncio_task import create_task
 from pymongo.read_preferences import ReadPreference
 
 _IS_SYNC = False
@@ -140,7 +139,7 @@ def __init__(self, **kwargs):
     if not _IS_SYNC:
 
         async def start(self):
-            self.task = create_task(self.run(), name=self.name)
+            self.task = asyncio.create_task(self.run(), name=self.name)
 
         async def join(self, timeout: Optional[float] = None):  # type: ignore[override]
             if self.task is not None:

test/asynchronous/test_client.py

@@ -1678,17 +1678,15 @@ async def test_contextlib(self):
         # pool
         self.assertEqual(1, len((await async_get_pool(client)).conns))
 
-        # contextlib async support was added in Python 3.10
-        if _IS_SYNC or sys.version_info >= (3, 10):
-            async with contextlib.aclosing(client):
-                self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                await client.pymongo_test.coll.find_one()
-            client = await self.async_rs_or_single_client()
-            async with client as client:
-                self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                await client.pymongo_test.coll.find_one()
+        async with contextlib.aclosing(client):
+            self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            await client.pymongo_test.coll.find_one()
+        client = await self.async_rs_or_single_client()
+        async with client as client:
+            self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            await client.pymongo_test.coll.find_one()
 
     @async_client_context.require_sync
     def test_interrupt_signal(self):

test/asynchronous/test_discovery_and_monitoring.py

@@ -327,8 +327,6 @@ class TestIgnoreStaleErrors(AsyncIntegrationTest):
         "PYTHON-5861: asyncio.Barrier hangs on macOS ARM64 CI",
     )
     async def test_ignore_stale_connection_errors(self):
-        if not _IS_SYNC and sys.version_info < (3, 11):
-            self.skipTest("Test requires asyncio.Barrier (added in Python 3.11)")
         N_TASKS = 5
         barrier = async_create_barrier(N_TASKS)
         client = await self.async_rs_or_single_client(minPoolSize=N_TASKS)

test/asynchronous/test_encryption.py

@@ -2053,14 +2053,11 @@ async def asyncSetUp(self):
         )
         # Errors when client has no cert, some examples:
         # [SSL: TLSV13_ALERT_CERTIFICATE_REQUIRED] tlsv13 alert certificate required (_ssl.c:2623)
+        # EOF occurred in violation of protocol (_ssl.c:2384)
         self.cert_error = (
             "certificate required|SSL handshake failed|"
-            "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE"
+            "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE|EOF"
         )
-        # On Python 3.10+ this error might be:
-        # EOF occurred in violation of protocol (_ssl.c:2384)
-        if sys.version_info[:2] >= (3, 10):
-            self.cert_error += "|EOF"
         # On Windows this error might be:
         # [WinError 10054] An existing connection was forcibly closed by the remote host
         if sys.platform == "win32":

test/asynchronous/test_locks.py

@@ -299,9 +299,6 @@ async def test_timeout_in_block(self):
                 with self.assertRaises(asyncio.TimeoutError):
                     await asyncio.wait_for(condition.wait(), timeout=0.5)
 
-        @unittest.skipIf(
-            sys.version_info < (3, 11), "raising the same cancelled error requires Python>=3.11"
-        )
         async def test_cancelled_error_wakeup(self):
             # Test that a cancelled error, received when awaiting wakeup,
             # will be re-raised un-modified.
@@ -328,9 +325,6 @@ async def func():
             # originally raised.
             self.assertIs(err.exception, raised)
 
-        @unittest.skipIf(
-            sys.version_info < (3, 11), "raising the same cancelled error requires Python>=3.11"
-        )
         async def test_cancelled_error_re_aquire(self):
             # Test that a cancelled error, received when re-aquiring lock,
             # will be re-raised un-modified.
@@ -363,7 +357,6 @@ async def func():
             # originally raised.
             self.assertIs(err.exception, raised)
 
-        @unittest.skipIf(sys.version_info < (3, 11), "asyncio.timeout requires Python>=3.11")
         async def test_cancelled_wakeup(self):
             # Test that a task cancelled at the "same" time as it is woken
             # up as part of a Condition.notify() does not result in a lost wakeup.
@@ -409,7 +402,6 @@ async def consumer():
                 condition.notify_all()
             await c[1]
 
-        @unittest.skipIf(sys.version_info < (3, 11), "asyncio.timeout requires Python>=3.11")
         async def test_cancelled_wakeup_relock(self):
             # Test that a task cancelled at the "same" time as it is woken
             # up as part of a Condition.notify() does not result in a lost wakeup.

test/asynchronous/test_retryable_writes.py

@@ -23,7 +23,6 @@
 import threading
 from unittest import mock
 
-from pymongo._asyncio_task import create_task
 from pymongo.common import MAX_ADAPTIVE_RETRIES
 from test.asynchronous.utils import async_set_fail_point, flaky
 
@@ -104,7 +103,7 @@ def succeeded(self, event: CommandSucceededEvent) -> None:
             else:
                 # succeeded() cannot await, so the fail point may be configured after
                 # the driver dispatches the retry it is meant to fail.
-                self._task = create_task(async_client_context.client.admin.command(cmd))  # type: ignore[arg-type]
+                self._task = asyncio.create_task(async_client_context.client.admin.command(cmd))  # type: ignore[arg-type]
                 self._task.add_done_callback(self._store_fail_point_exc)
 
 

test/asynchronous/test_ssl.py

@@ -189,10 +189,7 @@ class FakeSSLSock:
 
         self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Tests async sslobject_class injection (Python 3.11+ only)",
-    )
+    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection only")
     def test_async_tls_session_injected_via_sslobject_class(self):
         """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         fake_session = object()
@@ -260,10 +257,7 @@ async def run():
         asyncio.run(run())
         self.assertIs(cache[0], fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Tests async session injection on Python 3.11+",
-    )
+    @unittest.skipUnless(not _IS_SYNC, "Tests async session injection only")
     def test_async_configured_protocol_injects_session_via_sslobject_class(self):
         """When the cache has a session, sslobject_class is set and its __init__ body runs."""
         initial_session = object()

test/helpers.py

@@ -24,7 +24,6 @@
 
 from bson import SON
 from pymongo import common
-from pymongo._asyncio_task import create_task
 from pymongo.read_preferences import ReadPreference
 
 _IS_SYNC = True
@@ -140,7 +139,7 @@ def __init__(self, **kwargs):
     if not _IS_SYNC:
 
         def start(self):
-            self.task = create_task(self.run(), name=self.name)
+            self.task = asyncio.create_task(self.run(), name=self.name)
 
         def join(self, timeout: Optional[float] = None):  # type: ignore[override]
             if self.task is not None:

test/test_client.py

@@ -1633,17 +1633,15 @@ def test_contextlib(self):
         # pool
         self.assertEqual(1, len((get_pool(client)).conns))
 
-        # contextlib async support was added in Python 3.10
-        if _IS_SYNC or sys.version_info >= (3, 10):
-            with contextlib.closing(client):
-                self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                client.pymongo_test.coll.find_one()
-            client = self.rs_or_single_client()
-            with client as client:
-                self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                client.pymongo_test.coll.find_one()
+        with contextlib.closing(client):
+            self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            client.pymongo_test.coll.find_one()
+        client = self.rs_or_single_client()
+        with client as client:
+            self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            client.pymongo_test.coll.find_one()
 
     @client_context.require_sync
     def test_interrupt_signal(self):

test/test_discovery_and_monitoring.py

@@ -327,8 +327,6 @@ class TestIgnoreStaleErrors(IntegrationTest):
         "PYTHON-5861: asyncio.Barrier hangs on macOS ARM64 CI",
     )
     def test_ignore_stale_connection_errors(self):
-        if not _IS_SYNC and sys.version_info < (3, 11):
-            self.skipTest("Test requires asyncio.Barrier (added in Python 3.11)")
         N_TASKS = 5
         barrier = create_barrier(N_TASKS)
         client = self.rs_or_single_client(minPoolSize=N_TASKS)