Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Merge pull request #19 from akheron/database-info.

Add a method for requesting database info
  • Loading branch information...
commit 8706ff2706684e175a38e5073e5821b9f8c8d326 2 parents 00c24ec + 6f1501e
@nailor nailor authored
Showing with 44 additions and 0 deletions.
  1. +7 −0 doc/python-api.rst
  2. +21 −0 test/test_client.py
  3. +16 −0 trombi/client.py
View
7 doc/python-api.rst
@@ -244,6 +244,13 @@ argument.
as they are created via :meth:`Server.create` and
:meth:`Server.get`. Subclass of :class:`TrombiObject`.
+ .. method:: info(callback)
+
+ Request database information. Calls callback with a dict that
+ contains the info (see `here`__ for the dict contents).
+
+ __ http://techzone.couchbase.com/sites/default/files/uploads/all/documentation/couchbase-api-db.html#couchbase-api-db_db_get
+
.. method:: set([doc_id, ]data, callback[, attachments=None])
Creates a new or modifies an existing document in the database.
View
21 test/test_client.py
@@ -252,6 +252,27 @@ def callback(result):
@with_ioloop
@with_couchdb
+def test_db_info(baseurl, ioloop):
+ s = trombi.Server(baseurl, io_loop=ioloop)
+
+ def get_info(db):
+ print 'heere!'
+ db.info(check_info)
+
+ def check_info(info):
+ eq(info.error, False)
+ eq(info['db_name'], 'testdb')
+ eq(info['doc_count'], 0)
+ assert 'update_seq' in info
+ assert 'disk_size' in info
+ ioloop.stop()
+
+ s.create('testdb', callback=get_info)
+ ioloop.start()
+
+
+@with_ioloop
+@with_couchdb
def test_create_document(baseurl, ioloop):
def create_db_callback(db):
db.set(
View
16 trombi/client.py
@@ -99,6 +99,13 @@ def __init__(self, data):
super(TrombiResult, self).__init__()
+class TrombiDict(dict):
+ error = False
+
+ def to_basetype(self):
+ return dict(self)
+
+
def _jsonize_params(params):
result = dict()
for key, value in params.iteritems():
@@ -254,6 +261,15 @@ def _fetch(self, url, *args, **kwargs):
url = '%s/%s' % (self.baseurl, url)
return self.server._fetch(url, *args, **kwargs)
+ def info(self, callback):
+ def _really_callback(response):
+ if response.code == 200:
+ callback(TrombiDict(json.loads(response.body)))
+ else:
+ callback(_error_response(response))
+
+ self._fetch('', _really_callback)
+
def set(self, *args, **kwargs):
if len(args) == 2:
data, callback = args
Please sign in to comment.
Something went wrong with that request. Please try again.