Source code for wsgidav.samples.mongo_dav_provider

# -*- coding: utf-8 -*-
# (c) 2009-2023 Martin Wendt and contributors; see WsgiDAV
# Licensed under the MIT license:
Implementation of a WebDAV provider that provides a very basic, read-only
resource layer emulation of a MongoDB database.

Usage: add the following entries to wsgidav.conf::

    from wsgidav.samples.mongo_dav_provider import MongoResourceProvider
    mongo_dav_opts = {}
    addShare("mongo", MongoResourceProvider(mongo_dav_opts))

Valid options are (sample shows defaults)::

    opts = {"host": "localhost",       # MongoDB server
            "port": 27017,             # MongoDB port
            # This options are used with `mongod --auth`
            # The user must be created in the admin db with
            # > use admin
            # > db.addUser(user_name, password)
            "user": None,              # Authenticate with this user
            "pwd": None,               # ... and password

from io import StringIO
from pprint import pformat

import pymongo
from bson.objectid import ObjectId

from wsgidav import util
from wsgidav.dav_provider import DAVCollection, DAVNonCollection, DAVProvider
from wsgidav.util import join_uri

__docformat__ = "reStructuredText"

_logger = util.get_module_logger(__name__)

# ============================================================================
# ============================================================================
[docs] class ConnectionCollection(DAVCollection): """Root collection, lists all mongo databases.""" def __init__(self, path, environ): super().__init__(path, environ) self.conn = self.provider.conn
[docs] def get_member_names(self): return [name.encode("utf8") for name in self.conn.database_names()]
[docs] def get_member(self, name): return DbCollection(join_uri(self.path, name), self.environ)
[docs] class DbCollection(DAVCollection): """Mongo database, contains mongo collections.""" def __init__(self, path, environ): super().__init__(path, environ) self.conn = self.provider.conn self.db = self.conn[]
[docs] def get_display_info(self): return {"type": "Mongo database"}
[docs] def get_member_names(self): return [name.encode("utf8") for name in self.db.collection_names()]
[docs] def get_member(self, name): coll = self.db[name] return CollCollection(join_uri(self.path, name), self.environ, coll)
[docs] class CollCollection(DAVCollection): """Mongo collections, contains mongo documents.""" def __init__(self, path, environ, coll): super().__init__(path, environ) self.conn = self.provider.conn self.coll = coll
[docs] def get_display_info(self): return {"type": "Mongo collection"}
[docs] def get_member_names(self): res = [] for doc in self.coll.find(): res.append(util.to_str(doc["_id"])) return res
[docs] def get_member(self, name): doc = self.coll.find_one(ObjectId(name)) return DocResource(join_uri(self.path, name), self.environ, doc)
[docs] class DocResource(DAVNonCollection): """Mongo document, returned as virtual text resource.""" def __init__(self, path, environ, doc): super().__init__(path, environ) self.doc = doc
[docs] def get_content(self): html = "<pre>" + pformat(self.doc) + "</pre>" return StringIO(html.encode("utf8"))
[docs] def get_content_length(self): return len(self.get_content().read())
[docs] def get_content_type(self): return "text/html"
[docs] def get_display_name(self): doc = self.doc if doc.get("_title"): return doc["_title"].encode("utf8") elif doc.get("title"): return doc["title"].encode("utf8") elif doc.get("_id"): return util.to_str(doc["_id"]) return util.to_str(doc["key"])
[docs] def get_display_info(self): return {"type": "Mongo document"}
# ============================================================================ # MongoResourceProvider # ============================================================================
[docs] class MongoResourceProvider(DAVProvider): """DAV provider that serves a MongoDB structure.""" def __init__(self, options): super().__init__() self.options = options self.conn = pymongo.Connection(options.get("host"), options.get("port")) if options.get("user"): # If credentials are passed, acquire root access db = self.conn["admin"] res = db.authenticate(options.get("user"), options.get("pwd")) if not res: raise RuntimeError( "Failed to logon to db %s as user %s" % (, options.get("user")) ) "Logged on to mongo db '%s' as user '%s'" % (, options.get("user")) )"MongoResourceProvider connected to %s" % self.conn)
[docs] def get_resource_inst(self, path, environ): """Return DAVResource object for path. See DAVProvider.get_resource_inst() """"get_resource_inst('%s')" % path) self._count_get_resource_inst += 1 root = ConnectionCollection("/", environ) return root.resolve("/", path)