Source code for wsgidav.prop_man.couch_property_manager

# (c) 2009-2024 Martin Wendt and contributors; see WsgiDAV https://github.com/mar10/wsgidav
# Licensed under the MIT license: http://www.opensource.org/licenses/mit-license.php
"""
Implements a property manager based on CouchDB.


http://wiki.apache.org/couchdb/Reference
http://packages.python.org/CouchDB/views.html


Usage: add this lines to wsgidav.conf::

    from wsgidav.prop_man.couch_property_manager import CouchPropertyManager
    prop_man_opts = {}
    property_manager = CouchPropertyManager(prop_man_opts)

Valid options are (sample shows defaults)::

    opts = {"url": "http://localhost:5984/",  # CouchDB server
            "dbName": "wsgidav-props",        # Name of DB to store the properties
            }

"""
from urllib.parse import quote
from uuid import uuid4

import couchdb

from wsgidav import util

__docformat__ = "reStructuredText"

_logger = util.get_module_logger(__name__)

# ============================================================================
# CouchPropertyManager
# ============================================================================


[docs] class CouchPropertyManager: """Implements a property manager based on CouchDB.""" def __init__(self, options): self.options = options self._connect() def __del__(self): self._disconnect() def _connect(self): opts = self.options if opts.get("url"): self.couch = couchdb.Server(opts.get("url")) else: self.couch = couchdb.Server() dbName = opts.get("dbName", "wsgidav_props") if dbName in self.couch: self.db = self.couch[dbName] _logger.info( "CouchPropertyManager connected to %s v%s" % (self.db, self.couch.version()) ) else: self.db = self.couch.create(dbName) _logger.info( "CouchPropertyManager created new db %s v%s" % (self.db, self.couch.version()) ) # Ensure that we have a permanent view if "_design/properties" not in self.db: map = """ function(doc) { if(doc.type == 'properties') { emit(doc.url, { 'id': doc._id, 'url': doc.url }); } } """ designDoc = { "_id": "_design/properties", # "_rev": "42351258", "language": "javascript", "views": { "titles": { "map": ( "function(doc) { emit(null, { 'id': doc._id, " "'title': doc.title }); }" ) }, # http://127.0.0.1:5984/wsgidav_props/_design/properties/_view/by_url "by_url": {"map": map}, }, } self.db.save(designDoc) # pprint(self.couch.stats()) def _disconnect(self): pass def __repr__(self): return "CouchPropertyManager(%s)" % self.db def _sync(self): pass def _check(self, msg=""): pass def _dump(self, msg="", out=None): pass def _find(self, url): """Return properties document for path.""" # Query the permanent view to find a url vr = self.db.view("properties/by_url", key=url, include_docs=True) _logger.debug(f"find({url!r}) returned {len(vr)}") assert len(vr) <= 1, "Found multiple matches for %r" % url for row in vr: assert row.doc return row.doc return None def _find_descendents(self, url): """Return properties document for url and all children.""" # Ad-hoc query for URL starting with a prefix map_fun = """function(doc) { var url = doc.url + "/"; if(doc.type === 'properties' && url.indexOf('%s') === 0) { emit(doc.url, { 'id': doc._id, 'url': doc.url }); } }""" % ( url + "/" ) vr = self.db.query(map_fun, include_docs=True) for row in vr: yield row.doc return
[docs] def get_properties(self, norm_url, environ=None): _logger.debug("get_properties(%s)" % norm_url) doc = self._find(norm_url) propNames = [] if doc: for name in doc["properties"].keys(): propNames.append(name) return propNames
[docs] def get_property(self, norm_url, name, environ=None): _logger.debug(f"get_property({norm_url}, {name})") doc = self._find(norm_url) if not doc: return None prop = doc["properties"].get(name) return prop
[docs] def write_property( self, norm_url, name, property_value, dry_run=False, environ=None ): assert norm_url and norm_url.startswith("/") assert name assert property_value is not None _logger.debug( "write_property(%s, %s, dry_run=%s):\n\t%s" % (norm_url, name, dry_run, property_value) ) if dry_run: return # TODO: can we check anything here? doc = self._find(norm_url) if doc: doc["properties"][name] = property_value else: doc = { "_id": uuid4().hex, # Documentation suggests to set the id "url": norm_url, "title": quote(norm_url), "type": "properties", "properties": {name: property_value}, } self.db.save(doc)
[docs] def remove_property(self, norm_url, name, dry_run=False, environ=None): _logger.debug(f"remove_property({norm_url}, {name}, dry_run={dry_run})") if dry_run: # TODO: can we check anything here? return doc = self._find(norm_url) # Specifying the removal of a property that does not exist is NOT an error. if not doc or doc["properties"].get(name) is None: return del doc["properties"][name] self.db.save(doc)
[docs] def remove_properties(self, norm_url, environ=None): _logger.debug("remove_properties(%s)" % norm_url) doc = self._find(norm_url) if doc: self.db.delete(doc) return
[docs] def copy_properties(self, srcUrl, destUrl, environ=None): doc = self._find(srcUrl) if not doc: _logger.debug( f"copy_properties({srcUrl}, {destUrl}): src has no properties" ) return _logger.debug(f"copy_properties({srcUrl}, {destUrl})") assert not self._find(destUrl) doc2 = { "_id": uuid4().hex, "url": destUrl, "title": quote(destUrl), "type": "properties", "properties": doc["properties"], } self.db.save(doc2)
[docs] def move_properties(self, srcUrl, destUrl, with_children, environ=None): _logger.debug(f"move_properties({srcUrl}, {destUrl}, {with_children})") if with_children: # Match URLs that are equal to <srcUrl> or begin with '<srcUrl>/' docList = self._find_descendents(srcUrl) for doc in docList: newDest = doc["url"].replace(srcUrl, destUrl) _logger.debug("move property {} -> {}".format(doc["url"], newDest)) doc["url"] = newDest self.db.save(doc) else: # Move srcUrl only # TODO: use findAndModify()? doc = self._find(srcUrl) if doc: _logger.debug("move property {} -> {}".format(doc["url"], destUrl)) doc["url"] = destUrl self.db.save(doc) return
# ============================================================================ # # ============================================================================
[docs] def test(): pass
if __name__ == "__main__": test()