PYTHON-2707 Eliminate the use of 'slave'

Author: behackett

pymongo/aggregation.py

@@ -93,17 +93,18 @@ def _check_compat(sock_info):
         """Check whether the server version in-use supports aggregation."""
         pass
 
-    def _process_result(self, result, session, server, sock_info, slave_ok):
+    def _process_result(
+            self, result, session, server, sock_info, secondary_ok):
         if self._result_processor:
             self._result_processor(
-                result, session, server, sock_info, slave_ok)
+                result, session, server, sock_info, secondary_ok)
 
     def get_read_preference(self, session):
         if self._performs_write:
             return ReadPreference.PRIMARY
         return self._target._read_preference_for(session)
 
-    def get_cursor(self, session, server, sock_info, slave_ok):
+    def get_cursor(self, session, server, sock_info, secondary_ok):
         # Ensure command compatibility.
         self._check_compat(sock_info)
 
@@ -136,7 +137,7 @@ def get_cursor(self, session, server, sock_info, slave_ok):
         result = sock_info.command(
             self._database.name,
             cmd,
-            slave_ok,
+            secondary_ok,
             self.get_read_preference(session),
             self._target.codec_options,
             parse_write_concern_error=True,
@@ -147,7 +148,7 @@ def get_cursor(self, session, server, sock_info, slave_ok):
             client=self._database.client,
             user_fields=self._user_fields)
 
-        self._process_result(result, session, server, sock_info, slave_ok)
+        self._process_result(result, session, server, sock_info, secondary_ok)
 
         # Extract cursor from result or mock/fake one if necessary.
         if 'cursor' in result:

pymongo/change_stream.py

@@ -148,7 +148,8 @@ def _aggregation_pipeline(self):
         full_pipeline.extend(self._pipeline)
         return full_pipeline
 
-    def _process_result(self, result, session, server, sock_info, slave_ok):
+    def _process_result(
+            self, result, session, server, sock_info, secondary_ok):
         """Callback that caches the postBatchResumeToken or
         startAtOperationTime from a changeStream aggregate command response
         containing an empty batch of change documents.

pymongo/collection.py

@@ -197,7 +197,7 @@ def _socket_for_reads(self, session):
     def _socket_for_writes(self, session):
         return self.__database.client._socket_for_writes(session)
 
-    def _command(self, sock_info, command, slave_ok=False,
+    def _command(self, sock_info, command, secondary_ok=False,
                  read_preference=None,
                  codec_options=None, check=True, allowable_errors=None,
                  read_concern=None,
@@ -211,7 +211,7 @@ def _command(self, sock_info, command, slave_ok=False,
         :Parameters:
           - `sock_info` - A SocketInfo instance.
           - `command` - The command itself, as a SON instance.
-          - `slave_ok`: whether to set the SlaveOkay wire protocol bit.
+          - `secondary_ok`: whether to set the secondaryOkay wire protocol bit.
           - `codec_options` (optional) - An instance of
             :class:`~bson.codec_options.CodecOptions`.
           - `check`: raise OperationFailure if there are errors
@@ -238,7 +238,7 @@ def _command(self, sock_info, command, slave_ok=False,
             return sock_info.command(
                 self.__database.name,
                 command,
-                slave_ok,
+                secondary_ok,
                 read_preference or self._read_preference_for(session),
                 codec_options or self.codec_options,
                 check,
@@ -1629,13 +1629,13 @@ def parallel_scan(self, num_cursors, session=None, **kwargs):
                    ('numCursors', num_cursors)])
         cmd.update(kwargs)
 
-        with self._socket_for_reads(session) as (sock_info, slave_ok):
+        with self._socket_for_reads(session) as (sock_info, secondary_ok):
             # We call sock_info.command here directly, instead of
             # calling self._command to avoid using an implicit session.
             result = sock_info.command(
                 self.__database.name,
                 cmd,
-                slave_ok,
+                secondary_ok,
                 self._read_preference_for(session),
                 self.codec_options,
                 read_concern=self.read_concern,
@@ -1651,14 +1651,14 @@ def parallel_scan(self, num_cursors, session=None, **kwargs):
 
         return cursors
 
-    def _count_cmd(self, session, sock_info, slave_ok, cmd, collation):
+    def _count_cmd(self, session, sock_info, secondary_ok, cmd, collation):
         """Internal count command helper."""
         # XXX: "ns missing" checks can be removed when we drop support for
         # MongoDB 3.0, see SERVER-17051.
         res = self._command(
             sock_info,
             cmd,
-            slave_ok,
+            secondary_ok,
             allowable_errors=["ns missing"],
             codec_options=self.__write_response_codec_options,
             read_concern=self.read_concern,
@@ -1672,20 +1672,20 @@ def _count(self, cmd, collation=None, session=None):
         """Internal count helper."""
         # XXX: "ns missing" checks can be removed when we drop support for
         # MongoDB 3.0, see SERVER-17051.
-        def _cmd(session, server, sock_info, slave_ok):
+        def _cmd(session, server, sock_info, secondary_ok):
             return self._count_cmd(
-                session, sock_info, slave_ok, cmd, collation)
+                session, sock_info, secondary_ok, cmd, collation)
 
         return self.__database.client._retryable_read(
             _cmd, self._read_preference_for(session), session)
 
     def _aggregate_one_result(
-            self, sock_info, slave_ok, cmd, collation, session):
+            self, sock_info, secondary_ok, cmd, collation, session):
         """Internal helper to run an aggregate that returns a single result."""
         result = self._command(
             sock_info,
             cmd,
-            slave_ok,
+            secondary_ok,
             allowable_errors=[26],  # Ignore NamespaceNotFound.
             codec_options=self.__write_response_codec_options,
             read_concern=self.read_concern,
@@ -1719,7 +1719,7 @@ def estimated_document_count(self, **kwargs):
             raise ConfigurationError(
                 'estimated_document_count does not support sessions')
 
-        def _cmd(session, server, sock_info, slave_ok):
+        def _cmd(session, server, sock_info, secondary_ok):
             if sock_info.max_wire_version >= 12:
                 # MongoDB 4.9+
                 pipeline = [
@@ -1731,15 +1731,15 @@ def _cmd(session, server, sock_info, slave_ok):
                            ('cursor', {})])
                 cmd.update(kwargs)
                 result = self._aggregate_one_result(
-                    sock_info, slave_ok, cmd, collation=None, session=session)
+                    sock_info, secondary_ok, cmd, collation=None, session=session)
                 if not result:
                     return 0
                 return int(result['n'])
             else:
                 # MongoDB < 4.9
                 cmd = SON([('count', self.__name)])
                 cmd.update(kwargs)
-                return self._count_cmd(None, sock_info, slave_ok, cmd, None)
+                return self._count_cmd(None, sock_info, secondary_ok, cmd, None)
 
         return self.__database.client._retryable_read(
             _cmd, self.read_preference, None)
@@ -1816,9 +1816,9 @@ def count_documents(self, filter, session=None, **kwargs):
         collation = validate_collation_or_none(kwargs.pop('collation', None))
         cmd.update(kwargs)
 
-        def _cmd(session, server, sock_info, slave_ok):
+        def _cmd(session, server, sock_info, secondary_ok):
             result = self._aggregate_one_result(
-                sock_info, slave_ok, cmd, collation, session)
+                sock_info, secondary_ok, cmd, collation, session)
             if not result:
                 return 0
             return result['n']
@@ -2294,12 +2294,12 @@ def list_indexes(self, session=None):
         read_pref = ((session and session._txn_read_preference())
                      or ReadPreference.PRIMARY)
 
-        def _cmd(session, server, sock_info, slave_ok):
+        def _cmd(session, server, sock_info, secondary_ok):
             cmd = SON([("listIndexes", self.__name), ("cursor", {})])
             if sock_info.max_wire_version > 2:
                 with self.__database.client._tmp_session(session, False) as s:
                     try:
-                        cursor = self._command(sock_info, cmd, slave_ok,
+                        cursor = self._command(sock_info, cmd, secondary_ok,
                                                read_pref,
                                                codec_options,
                                                session=s)["cursor"]
@@ -2315,7 +2315,7 @@ def _cmd(session, server, sock_info, slave_ok):
             else:
                 res = message._first_batch(
                     sock_info, self.__database.name, "system.indexes",
-                    {"ns": self.__full_name}, 0, slave_ok, codec_options,
+                    {"ns": self.__full_name}, 0, secondary_ok, codec_options,
                     read_pref, cmd,
                     self.database.client._event_listeners)
                 cursor = res["cursor"]
@@ -2676,8 +2676,8 @@ def group(self, key, condition, initial, reduce, finalize=None, **kwargs):
         collation = validate_collation_or_none(kwargs.pop('collation', None))
         cmd.update(kwargs)
 
-        with self._socket_for_reads(session=None) as (sock_info, slave_ok):
-            return self._command(sock_info, cmd, slave_ok,
+        with self._socket_for_reads(session=None) as (sock_info, secondary_ok):
+            return self._command(sock_info, cmd, secondary_ok,
                                  collation=collation,
                                  user_fields={'retval': 1})["retval"]
 
@@ -2780,9 +2780,9 @@ def distinct(self, key, filter=None, session=None, **kwargs):
             kwargs["query"] = filter
         collation = validate_collation_or_none(kwargs.pop('collation', None))
         cmd.update(kwargs)
-        def _cmd(session, server, sock_info, slave_ok):
+        def _cmd(session, server, sock_info, secondary_ok):
             return self._command(
-                sock_info, cmd, slave_ok, read_concern=self.read_concern,
+                sock_info, cmd, secondary_ok, read_concern=self.read_concern,
                 collation=collation, session=session,
                 user_fields={"values": 1})["values"]
 
@@ -2809,7 +2809,7 @@ def _map_reduce(self, map, reduce, out, session, read_pref, **kwargs):
                      or read_pref)
 
         with self.__database.client._socket_for_reads(read_pref, session) as (
-                sock_info, slave_ok):
+                sock_info, secondary_ok):
             if (sock_info.max_wire_version >= 4 and
                     ('readConcern' not in cmd) and
                     inline):
@@ -2822,7 +2822,7 @@ def _map_reduce(self, map, reduce, out, session, read_pref, **kwargs):
                 write_concern = None
 
             return self._command(
-                sock_info, cmd, slave_ok, read_pref,
+                sock_info, cmd, secondary_ok, read_pref,
                 read_concern=read_concern,
                 write_concern=write_concern,
                 collation=collation, session=session,

pymongo/cursor.py

@@ -71,7 +71,7 @@
 
 _QUERY_OPTIONS = {
     "tailable_cursor": 2,
-    "slave_okay": 4,
+    "secondary_okay": 4,
     "oplog_replay": 8,
     "no_timeout": 16,
     "await_data": 32,

pymongo/database.py

@@ -620,8 +620,9 @@ def watch(self, pipeline=None, full_document=None, resume_after=None,
             batch_size, collation, start_at_operation_time, session,
             start_after)
 
-    def _command(self, sock_info, command, slave_ok=False, value=1, check=True,
-                 allowable_errors=None, read_preference=ReadPreference.PRIMARY,
+    def _command(self, sock_info, command, secondary_ok=False, value=1,
+                 check=True, allowable_errors=None,
+                 read_preference=ReadPreference.PRIMARY,
                  codec_options=DEFAULT_CODEC_OPTIONS,
                  write_concern=None,
                  parse_write_concern_error=False, session=None, **kwargs):
@@ -634,7 +635,7 @@ def _command(self, sock_info, command, slave_ok=False, value=1, check=True,
             return sock_info.command(
                 self.__name,
                 command,
-                slave_ok,
+                secondary_ok,
                 read_preference,
                 codec_options,
                 check,
@@ -748,8 +749,8 @@ def command(self, command, value=1, check=True,
             read_preference = ((session and session._txn_read_preference())
                                or ReadPreference.PRIMARY)
         with self.__client._socket_for_reads(
-                read_preference, session) as (sock_info, slave_ok):
-            return self._command(sock_info, command, slave_ok, value,
+                read_preference, session) as (sock_info, secondary_ok):
+            return self._command(sock_info, command, secondary_ok, value,
                                  check, allowable_errors, read_preference,
                                  codec_options, session=session, **kwargs)
 
@@ -761,15 +762,15 @@ def _retryable_read_command(self, command, value=1, check=True,
             read_preference = ((session and session._txn_read_preference())
                                or ReadPreference.PRIMARY)
 
-        def _cmd(session, server, sock_info, slave_ok):
-            return self._command(sock_info, command, slave_ok, value,
+        def _cmd(session, server, sock_info, secondary_ok):
+            return self._command(sock_info, command, secondary_ok, value,
                                  check, allowable_errors, read_preference,
                                  codec_options, session=session, **kwargs)
 
         return self.__client._retryable_read(
             _cmd, read_preference, session)
 
-    def _list_collections(self, sock_info, slave_okay, session,
+    def _list_collections(self, sock_info, secondary_okay, session,
                           read_preference, **kwargs):
         """Internal listCollections helper."""
 
@@ -782,7 +783,7 @@ def _list_collections(self, sock_info, slave_okay, session,
             with self.__client._tmp_session(
                     session, close=False) as tmp_session:
                 cursor = self._command(
-                    sock_info, cmd, slave_okay,
+                    sock_info, cmd, secondary_okay,
                     read_preference=read_preference,
                     session=tmp_session)["cursor"]
                 cmd_cursor = CommandCursor(
@@ -804,7 +805,7 @@ def _list_collections(self, sock_info, slave_okay, session,
             cmd = SON([("aggregate", "system.namespaces"),
                        ("pipeline", pipeline),
                        ("cursor", kwargs.get("cursor", {}))])
-            cursor = self._command(sock_info, cmd, slave_okay)["cursor"]
+            cursor = self._command(sock_info, cmd, secondary_okay)["cursor"]
             cmd_cursor = CommandCursor(coll, cursor, sock_info.address)
         cmd_cursor._maybe_pin_connection(sock_info)
         return cmd_cursor
@@ -833,9 +834,9 @@ def list_collections(self, session=None, filter=None, **kwargs):
         read_pref = ((session and session._txn_read_preference())
                      or ReadPreference.PRIMARY)
 
-        def _cmd(session, server, sock_info, slave_okay):
+        def _cmd(session, server, sock_info, secondary_okay):
             return self._list_collections(
-                sock_info, slave_okay, session, read_preference=read_pref,
+                sock_info, secondary_okay, session, read_preference=read_pref,
                 **kwargs)
 
         return self.__client._retryable_read(

pymongo/message.py

@@ -106,7 +106,7 @@ def _maybe_add_read_preference(spec, read_preference):
     # problems with mongos versions that don't support read preferences. Also,
     # for maximum backwards compatibility, don't add $readPreference for
     # secondaryPreferred unless tags or maxStalenessSeconds are in use (setting
-    # the slaveOkay bit has the same effect).
+    # the secondaryOkay bit has the same effect).
     if mode and (
             mode != ReadPreference.SECONDARY_PREFERRED.mode or
             len(document) > 1):
@@ -333,10 +333,10 @@ def as_command(self, sock_info):
         self._as_command = cmd, self.db
         return self._as_command
 
-    def get_message(self, set_slave_ok, sock_info, use_cmd=False):
-        """Get a query message, possibly setting the slaveOk bit."""
-        if set_slave_ok:
-            # Set the slaveOk bit.
+    def get_message(self, set_secondary_ok, sock_info, use_cmd=False):
+        """Get a query message, possibly setting the secondaryOk bit."""
+        if set_secondary_ok:
+            # Set the secondaryOk bit.
             flags = self.flags | 4
         else:
             flags = self.flags
@@ -349,7 +349,7 @@ def get_message(self, set_slave_ok, sock_info, use_cmd=False):
             if sock_info.op_msg_enabled:
                 request_id, msg, size, _ = _op_msg(
                     0, spec, self.db, self.read_preference,
-                    set_slave_ok, False, self.codec_options,
+                    set_secondary_ok, False, self.codec_options,
                     ctx=sock_info.compression_context)
                 return request_id, msg, size
             ns = _UJOIN % (self.db, "$cmd")
@@ -717,13 +717,13 @@ def _op_msg_uncompressed(flags, command, identifier, docs, check_keys, opts):
     _op_msg_uncompressed = _cmessage._op_msg
 
 
-def _op_msg(flags, command, dbname, read_preference, slave_ok, check_keys,
+def _op_msg(flags, command, dbname, read_preference, secondary_ok, check_keys,
             opts, ctx=None):
     """Get a OP_MSG message."""
     command['$db'] = dbname
     # getMore commands do not send $readPreference.
     if read_preference is not None and "$readPreference" not in command:
-        if slave_ok and not read_preference.mode:
+        if secondary_ok and not read_preference.mode:
             command["$readPreference"] = (
                 ReadPreference.PRIMARY_PREFERRED.document)
         else:
@@ -1717,7 +1717,7 @@ def unpack(cls, msg):
 
 
 def _first_batch(sock_info, db, coll, query, ntoreturn,
-                 slave_ok, codec_options, read_preference, cmd, listeners):
+                 secondary_ok, codec_options, read_preference, cmd, listeners):
     """Simple query helper for retrieving a first (and possibly only) batch."""
     query = _Query(
         0, db, coll, 0, query, None, codec_options,
@@ -1729,7 +1729,7 @@ def _first_batch(sock_info, db, coll, query, ntoreturn,
     if publish:
         start = datetime.datetime.now()
 
-    request_id, msg, max_doc_size = query.get_message(slave_ok, sock_info)
+    request_id, msg, max_doc_size = query.get_message(secondary_ok, sock_info)
 
     if publish:
         encoding_duration = datetime.datetime.now() - start