Merge pull request #14 from smacintyre/master

Adding support querying slave nodes

asyncmongo/client.py

@@ -36,7 +36,7 @@ class Client(object):
       - `**kwargs`: passed to `connection.Connection`
           - `host`: hostname or ip of mongo host
           - `port`: port to connect to
-          - `slave_ok` (optional): is it okay to connect directly to and perform queries on a slave instance
+          - `slave_okay` (optional): is it okay to connect directly to and perform queries on a slave instance
           - `autoreconnect` (optional): auto reconnect on interface errors
     
     @returns a `Client` instance that wraps a `pool.ConnectionPool`

asyncmongo/connection.py

@@ -35,19 +35,16 @@ class Connection(object):
     :Parameters:
       - `host`: hostname or ip of mongo host
       - `port`: port to connect to
-      - `slave_ok` (optional): is it okay to connect directly to and perform queries on a slave instance
       - `autoreconnect` (optional): auto reconnect on interface errors
       
     """
-    def __init__(self, host, port, slave_ok=False, autoreconnect=True, pool=None):
+    def __init__(self, host, port, autoreconnect=True, pool=None):
         assert isinstance(host, (str, unicode))
         assert isinstance(port, int)
-        assert isinstance(slave_ok, bool)
         assert isinstance(autoreconnect, bool)
         assert pool
         self.__host = host
         self.__port = port
-        self.__slave_ok = slave_ok
         self.__stream = None
         self.__callback = None
         self.__alive = False
@@ -159,4 +156,3 @@ def _parse_response(self, response):
         # logging.info('response: %s' % response)
         callback(response)
 
-

asyncmongo/cursor.py

@@ -40,6 +40,7 @@ def __init__(self, dbname, collection, pool):
         self.__dbname = dbname
         self.__collection = collection
         self.__pool = pool
+        self.__slave_okay = False
 
     @property
     def full_collection_name(self):
@@ -267,7 +268,7 @@ def find_one(self, spec_or_id, **kwargs):
 
     def find(self, spec=None, fields=None, skip=0, limit=0,
                  timeout=True, snapshot=False, tailable=False, sort=None,
-                 max_scan=None,
+                 max_scan=None, slave_okay=False,
                  _must_use_master=False, _is_command=False,
                  callback=None):
         """Query the database.
@@ -322,6 +323,8 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
             :meth:`~pymongo.cursor.Cursor.sort` for details.
           - `max_scan` (optional): limit the number of documents
             examined when performing the query
+          - `slave_okay` (optional): is it okay to connect directly
+            to and perform queries on a slave instance
         
         .. mongodoc:: find
         """
@@ -361,6 +364,7 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
         self.__snapshot = snapshot
         self.__ordering = sort and helpers._index_document(sort) or None
         self.__max_scan = max_scan
+        self.__slave_okay = slave_okay
         self.__explain = False
         self.__hint = None
         # self.__as_class = as_class
@@ -394,7 +398,7 @@ def __query_options(self):
         options = 0
         if self.__tailable:
             options |= _QUERY_OPTIONS["tailable_cursor"]
-        if False: #self.__collection.database.connection.slave_okay:
+        if self.__slave_okay or self.__pool.slave_okay:
             options |= _QUERY_OPTIONS["slave_okay"]
         if not self.__timeout:
             options |= _QUERY_OPTIONS["no_timeout"]

asyncmongo/pool.py

@@ -66,12 +66,14 @@ def __init__(self,
                 maxconnections=0, 
                 maxusage=0, 
                 dbname=None, 
+                slave_okay=False, 
                 *args, **kwargs):
         assert isinstance(mincached, int)
         assert isinstance(maxcached, int)
         assert isinstance(maxconnections, int)
         assert isinstance(maxusage, int)
         assert isinstance(dbname, (str, unicode, None.__class__))
+        assert isinstance(slave_okay, bool)
         if mincached and maxcached:
             assert mincached <= maxcached
         if maxconnections:
@@ -85,6 +87,7 @@ def __init__(self,
         self._idle_cache = [] # the actual connections that can be used
         self._condition = Condition()
         self._dbname = dbname
+        self._slave_okay = slave_okay
         self._connections = 0
 
         # Establish an initial number of idle database connections:
@@ -153,4 +156,14 @@ def close(self):
         finally:
             self._condition.release()
 
+    def __get_slave_okay(self):
+        """Is it OK to perform queries on a secondary or slave?
+        """
+        return self._slave_okay
+
+    def __set_slave_okay(self, value):
+        """Property setter for slave_okay"""
+        assert isinstance(value, bool)
+        self._slave_okay = value
 
+    slave_okay = property(__get_slave_okay, __set_slave_okay)