Ensure connection before we insert PYTHON-414

Inserts require that we know the max_bson_size and max_message_size of the current primary or master. We have to be connected to get those values.

Author: behackett

pymongo/collection.py

@@ -341,6 +341,11 @@ def insert(self, doc_or_docs, manipulate=True,
 
  .. mongodoc:: insert
  """
+ # Batch inserts require us to know the connected master's
+ # max_bson_size and max_message_size. We have to be connected
+ # to a master to know that.
+ self.database.connection._ensure_connected(True)
+
  docs = doc_or_docs
  return_one = False
  if isinstance(docs, dict):

pymongo/master_slave_connection.py

@@ -159,6 +159,11 @@ def set_cursor_manager(self, manager_class):
  for slave in self.__slaves:
  slave.set_cursor_manager(manager_class)
 
+ def _ensure_connected(self, sync):
+ """Ensure the master is connected to a mongod/s.
+ """
+ self.__master._ensure_connected(sync)
+
  # _connection_to_use is a hack that we need to include to make sure
  # that killcursor operations can be sent to the same instance on which
  # the cursor actually resides...

pymongo/mongo_client.py

@@ -778,6 +778,13 @@ def __socket(self):
  raise
  return sock_info
 
+ def _ensure_connected(self, dummy):
+ """Ensure this client instance is connected to a mongod/s.
+ """
+ host, port = (self.__host, self.__port)
+ if host is None or (port is None and '/' not in host):
+ self.__find_node()
+
  def disconnect(self):
  """Disconnect from MongoDB.
 

pymongo/mongo_replica_set_client.py

@@ -1224,6 +1224,22 @@ def __socket(self, member, force=False):
  raise
  return sock_info
 
+ def _ensure_connected(self, sync=False):
+ """Ensure this client instance is connected to a primary.
+ """
+ # This may be the first time we're connecting to the set.
+ if self.__monitor and not self.__monitor.started:
+ try:
+ self.__monitor.start()
+ # Minor race condition. It's possible that two (or more)
+ # threads could call monitor.start() consecutively. Just pass.
+ except RunTimeError:
+ pass
+ if sync:
+ rs_state = self.__rs_state
+ if not rs_state.primary_member:
+ self.__schedule_refresh(sync)
+
  def disconnect(self):
  """Disconnect from the replica set primary, unpin all members, and
  refresh our view of the replica set.
@@ -1390,9 +1406,7 @@ def _send_message(self, msg,
  - `with_last_error`: check getLastError status after sending the
  message
  """
- # This may be the first time we're connecting to the set.
- if self.__monitor and not self.__monitor.started:
- self.__monitor.start()
+ self._ensure_connected()
 
  if _connection_to_use in (None, -1):
  member = self.__find_primary()
@@ -1494,9 +1508,7 @@ def _send_message_with_response(self, msg, _connection_to_use=None,
  used by Cursor for getMore and killCursors messages.
  - `_must_use_master`: If True, send to primary.
  """
- # This may be the first time we're connecting to the set.
- if self.__monitor and not self.__monitor.started:
- self.__monitor.start()
+ self._ensure_connected()
 
  rs_state = self.__rs_state
  tag_sets = kwargs.get('tag_sets', [{}])

test/test_replica_set_client.py

@@ -126,6 +126,14 @@ def test_init_disconnected(self):
  c.pymongo_test.test.update({}, {}) # Auto-connect for write.
  self.assertTrue(c.primary)
 
+ c = self._get_client(_connect=False)
+ c.pymongo_test.test.insert({}) # Auto-connect for write.
+ self.assertTrue(c.primary)
+
+ c = self._get_client(_connect=False)
+ c.pymongo_test.test.remove({}) # Auto-connect for write.
+ self.assertTrue(c.primary)
+
  c = MongoReplicaSetClient(
  "somedomainthatdoesntexist.org", replicaSet="rs",
  connectTimeoutMS=1, _connect=False)