PYTHON-6029 Consolidate non-IO code in Pool and Connection into shared base class

pymongo/_telemetry.py

@@ -37,12 +37,12 @@
     _verbose_connection_error_reason,
 )
 from pymongo.message import _randint
-from pymongo.pool_shared import _ConnectionTelemetryInfo
 
 if TYPE_CHECKING:
     from bson.objectid import ObjectId
     from pymongo.hello import Hello
     from pymongo.monitoring import _EventListeners
+    from pymongo.pool_shared import _ConnectionTelemetryInfo
     from pymongo.server_description import ServerDescription
     from pymongo.topology_description import TopologyDescription
     from pymongo.typings import _Address, _DocumentOut

pymongo/asynchronous/bulk.py

@@ -236,7 +236,7 @@ async def _execute_command(
                 conn.send_cluster_time(cmd, session, client)
                 conn.add_server_api(cmd)
                 # CSOT: apply timeout before encoding the command.
-                conn.apply_timeout(client, cmd)
+                conn.apply_timeout(cmd)
                 ops = islice(run.ops, run.idx_offset, None)
 
                 # Run as many ops as possible in one command.

pymongo/asynchronous/client_bulk.py

@@ -279,7 +279,7 @@ async def _execute_command(
             conn.send_cluster_time(cmd, session, self.client)
             conn.add_server_api(cmd)
             # CSOT: apply timeout before encoding the command.
-            conn.apply_timeout(self.client, cmd)
+            conn.apply_timeout(cmd)
             ops = islice(self.ops, self.idx_offset, None)
             namespaces = islice(self.namespaces, self.idx_offset, None)
 

pymongo/asynchronous/command_runner.py

@@ -422,7 +422,7 @@ async def run_command(
 
     # Support CSOT
     if client:
-        conn.apply_timeout(client, spec)
+        conn.apply_timeout(spec)
     _csot.apply_write_concern(spec, write_concern)
 
     flags = _OpMsg.MORE_TO_COME if unacknowledged else 0