# -*- test-case-name: couchdb.test.test_couchdb -*-# -*- coding: utf-8 -*-## Copyright (C) 2007-2008 Christopher Lenz# Copyright (C) 2008 Federico Stella# All rights reserved.## Redistribution and use in source and binary forms, with or without# modification, are permitted provided that the following conditions# are met:# # 1. Redistributions of source code must retain the above copyright# notice, this list of conditions and the following disclaimer.# 2. Redistributions in binary form must reproduce the above copyright# notice, this list of conditions and the following disclaimer in# the documentation and/or other materials provided with the# distribution.# 3. The name of the author may not be used to endorse or promote# products derived from this software without specific prior# written permission.## THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE# ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE# GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER# IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR# OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN# IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.import refrom twisted.internet import reactorfrom twisted.web.client import getPagefrom twisted.python.util import printlnfrom twisted.web import client, error as weberrorfrom twisted.python.failure import Failurefrom urllib import quote, urlencodefrom urlparse import urlparseimport simplejson as jsonimport sys__all__ = ['ResourceNotFound', 'ResourceConflict', 'ServerError', 'Server', 'Database', 'Document', 'ViewResults', 'Row']class ResourceNotFound(Exception): """Exception raised when a 404 HTTP error is received in response to a request. """class ResourceConflict(Exception): """Exception raised when a 409 HTTP error is received in response to a request. """class ServerError(Exception): """Exception raised when an unexpected HTTP error is received in response to a request. """VALID_DB_NAME = re.compile(r'^[a-z0-9_$()+-/]+$')def validate_dbname(name): if not VALID_DB_NAME.match(name): raise ValueError('Invalid database name') return namedef uri(base, *path, **query): """Assemble a uri based on a base, any number of path segments, and query string parameters. >>> uri('http://example.org/', '/_all_dbs') 'http://example.org/_all_dbs' """ if base and base.endswith('/'): base = base[:-1] retval = [base] # build the path path = '/'.join([''] + [unicode_quote(s.strip('/')) for s in path if s is not None]) if path: retval.append(path) # build the query string params = [] for name, value in query.items(): if type(value) in (list, tuple): params.extend([(name, i) for i in value if i is not None]) elif value is not None: params.append((name, value)) if params: retval.extend(['?', unicode_urlencode(params)]) return ''.join(retval)def unicode_quote(string, safe=''): if isinstance(string, unicode): string = string.encode('utf-8') return quote(string, safe)def unicode_urlencode(data): if isinstance(data, dict): data = data.items() params = [] for name, value in data: if isinstance(value, unicode): value = value.encode('utf-8') params.append((name, value)) return urlencode(params)def _parse(url, defaultPort=None): """ Split the given URL into the scheme, host, port, and path. @type url: C{str} @param url: An URL to parse. @type defaultPort: C{int} or C{None} @param defaultPort: An alternate value to use as the port if the URL does not include one. @return: A four-tuple of the scheme, host, port, and path of the URL. All of these are C{str} instances except for port, which is an C{int}. """ scheme, host, port, path = urlparse(url)[:4] if ':' in host: host, port = host.split(':') port = int(port) if path == "": path = "/" return scheme, host, port, pathdef rest_action(url, contextFactory=None, content=None, headers=None, *args, **kwargs): """Download a web page as a string. Download a page. Return a deferred, which will callback with a page (as a string) or errback with a description of the error. See HTTPClientFactory to see what extra args can be passed. """ headers = headers or {} headers.setdefault('Accept', 'application/json') body = None if content: if not isinstance(content, basestring): body = json.dumps(content) headers.setdefault('Content-Type', 'application/json') else: body = content scheme, host, port, path = _parse(url) factory = client.HTTPClientFactory(url, postdata=body, headers=headers, agent='twisted-couchdb', *args, **kwargs) if scheme == 'https': from twisted.internet import ssl if contextFactory is None: contextFactory = ssl.ClientContextFactory() reactor.connectSSL(host, port, factory, contextFactory) else: reactor.connectTCP(host, port, factory) return factory.deferredclass Resource(object): def __init__(self, uri): self.uri = uri def delete(self, path=None, headers=None, **params): return rest_action( uri(self.uri, path, **params), method='DELETE', headers=headers ).addCallbacks(self._to_json, self._raise_error) def get(self, path=None, headers=None, **params): return rest_action( uri(self.uri, path, **params), method='GET', headers=headers ).addCallbacks(self._to_json, self._raise_error) def head(self, path=None, headers=None, **params): return rest_action( uri(self.uri, path, **params), method='HEAD', headers=headers ).addCallbacks(self._to_json, self._raise_error) def post(self, path=None, content=None, headers=None, **params): return rest_action( uri(self.uri, path, **params), method='POST', content=content, headers=headers ).addCallbacks(self._to_json, self._raise_error) def put(self, path=None, content=None, headers=None, **params): return rest_action( uri(self.uri, path, **params), method='PUT', content=content, headers=headers ).addCallbacks(self._to_json, self._raise_error) def _to_json(self, data): try: return json.loads(data) except ValueError: return {} def _raise_error(self, failure): failure.trap(weberror.Error) status = int(failure.value.status) data = failure.value.response try: data = json.loads(data) except ValueError: pass if status >= 400: if isinstance(failure.value.response, dict): error = (data.get('error'), data.get('reason')) else: error = data if status == 404: return Failure(ResourceNotFound(error)) elif status == 409: return Failure(ResourceConflict(error)) else: return Failure(ServerError((status, error)))class Server(object): """Representation of a CouchDB server.""" def __init__(self, uri): """Initialize the server object. :param uri: the URI of the server (for example ``http://localhost:5984/``) """ self.resource = Resource(uri) self.uri = uri def __repr__(self): return '<%s %r>' % (type(self).__name__, self.resource.uri) def _get_version(self): return self.resource.get("").addCallback(lambda r: r['version']) def len(self): """Return the number of databases.""" return self.resource.get('_all_dbs').addCallback(len) def list(self): """List the names of all databases.""" return self.resource.get('_all_dbs') def create(self, name): """Create a new database with the given name.""" return self.resource.put(validate_dbname(name)) def delete(self, name): """Remove the database with the specified name.""" return self.resource.delete(validate_dbname(name)) def get(self, name): """Return a `Database` object representing the database with the specified name.""" return Database(uri(self.uri, validate_dbname(name))) def __getitem__(self, name): """Return a `Database` object representing the database with the specified name.""" return self.get(name)class Database(object): """Representation of a database on a CouchDB server.""" def __init__(self, uri, name=None): self.resource = Resource(uri) self.name = name def contains(self, id): """Return whether the database contains a document with the specified""" return self.resource.head(id).addCallbacks(lambda _: True, lambda _: False) def iter(self): """Return the IDs of all documents in the database.""" return self.view('_all_docs').addCallback(lambda r: iter([item.id for item in r])) def len(self): """Return the number of documents in the database.""" return self.resource.get().addCallback(lambda x: x['doc_count']) def get(self, id, default=None, **options): """Return the document with the specified ID.""" return self.resource.get(id).addCallbacks(lambda r: Document(r), lambda _: default) def set(self, id, content): """Create or update a document with the specified ID.""" def update(result): content.update({'_id': result['id'], '_rev': result['rev']}) return content return self.resource.put(id, content=content).addCallbacks(update) def create(self, content): """Create a new document in the database with a generated ID. Any keyword arguments are used to populate the fields of the new document. """ return self.resource.post(content=content).addCallback(lambda r: r['id']) def delete(self, id): """Delete the document with the specified ID.""" return self.get(id).addCallback(lambda r: self.resource.delete(id, rev=r.rev)) def update(self, documents): """Perform a bulk update or insertion of the given documents using a single HTTP request. """ documents = list(documents) def cb(retval): for idx, result in enumerate(retval['results']): assert 'ok' in result # FIXME: how should error handling work here? doc = documents[idx] doc.update({'_id': result['id'], '_rev': result['rev']}) yield doc return self.resource.post('_bulk_docs', content=documents).addCallback(cb) def view(self, name, wrapper=None, **options): """Execute a predefined view.""" view = PermanentView(uri(self.resource.uri, *name.split('/')), name, wrapper=wrapper) return view(**options) def query(self, code, content_type='text/javascript', wrapper=None, **options): """Execute an ad-hoc query (a "temp view") against the database.""" view = TemporaryView(uri(self.resource.uri, '_temp_view'), code, content_type=content_type, wrapper=wrapper) return view(**options)class Document(dict): """Representation of a document in the database. This is basically just a dictionary with the two additional properties `id` and `rev`, which contain the document ID and revision, respectively. """ def __init__(self, *args, **kwargs): dict.__init__(self, *args, **kwargs) def __repr__(self): return '<%s %r@%r %r>' % (type(self).__name__, self.id, self.rev, dict([(k,v) for k,v in self.items() if k not in ('_id', '_rev')]) ) id = property(lambda self: self['_id']) rev = property(lambda self: self['_rev'])class View(object): """Abstract representation of a view or query.""" def __init__(self, uri, wrapper=None): self.resource = Resource(uri) self.wrapper = wrapper def __call__(self, **options): #return ViewResults(self, options) v = ViewResults(self, options) return v._fetch().addCallback(lambda _: v) def __iter__(self): return self() def _encode_options(self, options): retval = {} for name, value in options.items(): if name in ('key', 'startkey', 'endkey') or not isinstance(value, basestring): value = json.dumps(value) retval[name] = value return retval def _exec(self, options): raise NotImplementedErrorclass PermanentView(View): """Representation of a permanent view on the server.""" def __init__(self, uri, name, wrapper=None): View.__init__(self, uri, wrapper=wrapper) self.resource = Resource(uri) self.name = name def __repr__(self): return '<%s %r>' % (type(self).__name__, self.name) def _exec(self, options): return self.resource.get(**self._encode_options(options))class TemporaryView(View): """Representation of a temporary view.""" def __init__(self, uri, code=None, content_type='text/javascript', wrapper=None): View.__init__(self, uri, wrapper=wrapper) self.resource = Resource(uri) self.code = code self.content_type = content_type def __repr__(self): return '<%s %r>' % (type(self).__name__, self.code) def _exec(self, options): headers = {} if self.content_type: headers['Content-Type'] = self.content_type return self.resource.post(content=self.code, headers=headers, **self._encode_options(options))class ViewResults(object): """Representation of a parameterized view (either permanent or temporary) and the results it produces. This class allows the specification of ``key``, ``startkey``, and ``endkey`` options using Python slice notation. Use plain indexed notation (without a slice) to apply the ``key`` option. Note that as CouchDB makes no claim that keys are unique in a view, this can still return multiple rows: """ def __init__(self, view, options): self.view = view self.options = options self._rows = self._total_rows = self._offset = None def __repr__(self): return '<%s %r %r>' % (type(self).__name__, self.view, self.options) def __getitem__(self, key): options = self.options.copy() if type(key) is slice: if key.start is not None: options['startkey'] = key.start if key.stop is not None: options['endkey'] = key.stop return ViewResults(self.view, options) else: options['key'] = key return ViewResults(self.view, options) def __iter__(self): wrapper = self.view.wrapper for row in self.rows: if wrapper is not None: yield wrapper(row) else: yield row def __len__(self): return len(self.rows) def _fetch(self): def cb(data): self._rows = [Row(r['id'], r['key'], r['value']) for r in data['rows']] self._total_rows = data['total_rows'] self._offset = data.get('offset', 0) return self.view._exec(self.options).addCallback(cb) def _get_rows(self): if self._rows is None: self._fetch() return self._rows rows = property(_get_rows, doc="""\ The list of rows returned by the view. :type: `list` """) def _get_total_rows(self): if self._total_rows is None: self._fetch() return self._total_rows total_rows = property(_get_total_rows, doc="""\ The total number of rows in this view. :type: `int` """) def _get_offset(self): if self._offset is None: self._fetch() return self._offset offset = property(_get_offset, doc="""\ The offset of the results from the first row in the view. :type: `int` """)class Row(object): """Representation of a row as returned by database views. This is basically just a dictionary with the two additional properties `id` and `rev`, which contain the document ID and revision, respectively. """ def __init__(self, id, key, value): self.id = id self.key = key self.value = value def __repr__(self): return '<%s id=%r, key=%r, value=%r>' % (type(self).__name__, self.id, self.key, self.value)