mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into PYTHON-5708

Browse Source
This commit is contained in:
Noah Stapp 2026-05-07 12:28:33 -04:00 committed by GitHub
commit 279fbdcd7d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 128 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
CONTRIBUTING.md
View File

@ -505,13 +505,20 @@ python3 ./.evergreen/scripts/resync-all-specs.py
Follow the [Python Driver Release Process Wiki](https://wiki.corp.mongodb.com/display/DRIVERS/Python+Driver+Release+Process).
## Asyncio considerations
## Project Structure and Asyncio Considerations
PyMongo adds asyncio capability by modifying the source files in `*/asynchronous` to `*/synchronous` using
[unasync](https://github.com/python-trio/unasync/) and some custom transforms.
This section describes the layout of the `pymongo/` package.
Where possible, edit the code in `*/asynchronous/*.py` and not the synchronous files.
You can run `pre-commit run --all-files synchro` before running tests if you are testing synchronous code.
Within `pymongo/`, the code is further divided into the `pymongo/asynchronous` and `pymongo/synchronous` subdirectories.
Files in `pymongo/synchronous` are generated from `pymongo/asynchronous` using the `synchro` pre-commit hook, which uses [unasync](https://github.com/python-trio/unasync/) and some custom transforms.
As a result, **all modifications** within `pymongo` must be made in either the top-level `pymongo` directory when they have to exhibit differing behavior between sync and async contexts or the `pymongo/asynchronous` directory, not `pymongo/synchronous`.
Any changes made directly to files in the `pymongo/synchronous` directory will be overwritten by the `synchro` hook when it is run, which happens automatically on commit.
Some top-level files (e.g. `pymongo/collection.py`) are re-export files for existing import compatibility and should not be modified directly.
The other top-level files (e.g. `pymongo/network_layer.py`, `pymongo/pool_shared.py`) contain either shared code used in both the asynchronous and synchronous APIs, or code that is very different between the two APIs and therefore cannot be generated from the async version using `synchro`.
Run `pre-commit run --all-files synchro` before running tests to generate the latest version of the synchronous code.
To prevent the `synchro` hook from accidentally overwriting code, it first checks to see whether a sync version
of a file is changing and not its async counterpart, and will fail.

6
pymongo/asynchronous/client_session.py
View File

@ -1101,7 +1101,11 @@ class AsyncClientSession:
read_preference: _ServerMode,
conn: AsyncConnection,
) -> None:
if not conn.supports_sessions:
# getMores must be sent with a session if the cursor was opened with one
operation = next(iter(command))
if not conn.supports_sessions and (
isinstance(self._server_session, _EmptyServerSession) or operation != "getMore"
):
if not self._implicit:
raise ConfigurationError("Sessions are not supported by this MongoDB deployment")
return

37
pymongo/asynchronous/pool.py
View File

@ -760,11 +760,7 @@ class Pool:
self._pending = 0
self._max_connecting = self.opts.max_connecting
self._client_id = client_id
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_created(
self.address, self.opts.non_default_options
)
# Log before publishing event to prevent potential listener preemption in tests
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -774,6 +770,11 @@ class Pool:
serverPort=self.address[1],
**self.opts.non_default_options,
)
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_created(
self.address, self.opts.non_default_options
)
# Similar to active_sockets but includes threads in the wait queue.
self.operation_count: int = 0
# Retain references to pinned connections to prevent the CPython GC
@ -788,9 +789,6 @@ class Pool:
async with self.lock:
if self.state != PoolState.READY:
self.state = PoolState.READY
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_ready(self.address)
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -799,6 +797,9 @@ class Pool:
serverHost=self.address[0],
serverPort=self.address[1],
)
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_ready(self.address)
@property
def closed(self) -> bool:
@ -859,9 +860,6 @@ class Pool:
else:
for conn in sockets:
await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_closed(self.address)
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -870,15 +868,11 @@ class Pool:
serverHost=self.address[0],
serverPort=self.address[1],
)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_closed(self.address)
else:
if old_state != PoolState.PAUSED:
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_cleared(
self.address,
service_id=service_id,
interrupt_connections=interrupt_connections,
)
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -888,6 +882,13 @@ class Pool:
serverPort=self.address[1],
serviceId=service_id,
)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_cleared(
self.address,
service_id=service_id,
interrupt_connections=interrupt_connections,
)
if not _IS_SYNC:
await asyncio.gather(
*[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets], # type: ignore[func-returns-value]

6
pymongo/synchronous/client_session.py
View File

@ -1097,7 +1097,11 @@ class ClientSession:
read_preference: _ServerMode,
conn: Connection,
) -> None:
if not conn.supports_sessions:
# getMores must be sent with a session if the cursor was opened with one
operation = next(iter(command))
if not conn.supports_sessions and (
isinstance(self._server_session, _EmptyServerSession) or operation != "getMore"
):
if not self._implicit:
raise ConfigurationError("Sessions are not supported by this MongoDB deployment")
return

37
pymongo/synchronous/pool.py
View File

@ -758,11 +758,7 @@ class Pool:
self._pending = 0
self._max_connecting = self.opts.max_connecting
self._client_id = client_id
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_created(
self.address, self.opts.non_default_options
)
# Log before publishing event to prevent potential listener preemption in tests
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -772,6 +768,11 @@ class Pool:
serverPort=self.address[1],
**self.opts.non_default_options,
)
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_created(
self.address, self.opts.non_default_options
)
# Similar to active_sockets but includes threads in the wait queue.
self.operation_count: int = 0
# Retain references to pinned connections to prevent the CPython GC
@ -786,9 +787,6 @@ class Pool:
with self.lock:
if self.state != PoolState.READY:
self.state = PoolState.READY
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_ready(self.address)
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -797,6 +795,9 @@ class Pool:
serverHost=self.address[0],
serverPort=self.address[1],
)
if self.enabled_for_cmap:
assert self.opts._event_listeners is not None
self.opts._event_listeners.publish_pool_ready(self.address)
@property
def closed(self) -> bool:
@ -857,9 +858,6 @@ class Pool:
else:
for conn in sockets:
conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_closed(self.address)
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -868,15 +866,11 @@ class Pool:
serverHost=self.address[0],
serverPort=self.address[1],
)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_closed(self.address)
else:
if old_state != PoolState.PAUSED:
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_cleared(
self.address,
service_id=service_id,
interrupt_connections=interrupt_connections,
)
if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
_debug_log(
_CONNECTION_LOGGER,
@ -886,6 +880,13 @@ class Pool:
serverPort=self.address[1],
serviceId=service_id,
)
if self.enabled_for_cmap:
assert listeners is not None
listeners.publish_pool_cleared(
self.address,
service_id=service_id,
interrupt_connections=interrupt_connections,
)
if not _IS_SYNC:
asyncio.gather(
*[conn.close_conn(ConnectionClosedReason.STALE) for conn in sockets], # type: ignore[func-returns-value]

38
test/asynchronous/test_session.py
View File

@ -15,7 +15,6 @@
"""Test the client_session module."""
from __future__ import annotations
import asyncio
import copy
import sys
import time
@ -24,8 +23,6 @@ from io import BytesIO
from test.asynchronous.helpers import ExceptionCatchingTask
from typing import Any, Callable, List, Set, Tuple
from pymongo.synchronous.mongo_client import MongoClient
sys.path[0:0] = [""]
from test.asynchronous import (
@ -45,7 +42,7 @@ from test.utils_shared import (
from bson import DBRef
from gridfs.asynchronous.grid_file import AsyncGridFS, AsyncGridFSBucket
from pymongo import ASCENDING, AsyncMongoClient, _csot, monitoring
from pymongo import ASCENDING, AsyncMongoClient, monitoring
from pymongo.asynchronous.command_cursor import AsyncCommandCursor
from pymongo.asynchronous.cursor import AsyncCursor
from pymongo.asynchronous.helpers import anext
@ -938,6 +935,39 @@ class TestSession(AsyncIntegrationTest):
await s2.end_session()
async def test_getmore_preserves_lsid_after_session_support_lost(self):
listener = OvertCommandListener()
client = await self.async_rs_or_single_client(event_listeners=[listener], maxPoolSize=1)
coll = client.pymongo_test.test
await coll.drop()
await coll.insert_many([{"x": i} for i in range(10)])
self.addAsyncCleanup(coll.drop)
async with client.start_session() as s:
cursor = coll.find({}, batch_size=2, session=s)
await anext(cursor)
find_event = next(e for e in listener.started_events if e.command_name == "find")
lsid = find_event.command["lsid"]
# Simulate a node stepping down: mark idle connections as not supporting sessions.
for server in client._topology._servers.values():
for conn in server.pool.conns:
conn.supports_sessions = False
listener.reset()
await cursor.to_list()
getmore_events = [e for e in listener.started_events if e.command_name == "getMore"]
self.assertGreater(len(getmore_events), 0, "expected at least one getMore command")
for event in getmore_events:
self.assertIn(
"lsid", event.command, "getMore must include lsid when session is materialized"
)
self.assertEqual(
lsid, event.command["lsid"], "getMore lsid must match the session lsid from find"
)
class TestCausalConsistency(AsyncUnitTest):
listener: SessionTestListener

38
test/test_session.py
View File

@ -15,7 +15,6 @@
"""Test the client_session module."""
from __future__ import annotations
import asyncio
import copy
import sys
import time
@ -24,8 +23,6 @@ from io import BytesIO
from test.helpers import ExceptionCatchingTask
from typing import Any, Callable, List, Set, Tuple
from pymongo.synchronous.mongo_client import MongoClient
sys.path[0:0] = [""]
from test import (
@ -45,7 +42,7 @@ from test.utils_shared import (
from bson import DBRef
from gridfs.synchronous.grid_file import GridFS, GridFSBucket
from pymongo import ASCENDING, MongoClient, _csot, monitoring
from pymongo import ASCENDING, MongoClient, monitoring
from pymongo.common import _MAX_END_SESSIONS
from pymongo.errors import ConfigurationError, InvalidOperation, OperationFailure
from pymongo.operations import IndexModel, InsertOne, UpdateOne
@ -938,6 +935,39 @@ class TestSession(IntegrationTest):
s2.end_session()
def test_getmore_preserves_lsid_after_session_support_lost(self):
listener = OvertCommandListener()
client = self.rs_or_single_client(event_listeners=[listener], maxPoolSize=1)
coll = client.pymongo_test.test
coll.drop()
coll.insert_many([{"x": i} for i in range(10)])
self.addCleanup(coll.drop)
with client.start_session() as s:
cursor = coll.find({}, batch_size=2, session=s)
next(cursor)
find_event = next(e for e in listener.started_events if e.command_name == "find")
lsid = find_event.command["lsid"]
# Simulate a node stepping down: mark idle connections as not supporting sessions.
for server in client._topology._servers.values():
for conn in server.pool.conns:
conn.supports_sessions = False
listener.reset()
cursor.to_list()
getmore_events = [e for e in listener.started_events if e.command_name == "getMore"]
self.assertGreater(len(getmore_events), 0, "expected at least one getMore command")
for event in getmore_events:
self.assertIn(
"lsid", event.command, "getMore must include lsid when session is materialized"
)
self.assertEqual(
lsid, event.command["lsid"], "getMore lsid must match the session lsid from find"
)
class TestCausalConsistency(UnitTest):
listener: SessionTestListener