Custom datatypes in Ming

ming_ipaddress.py

from datetime import datetime
from socket import inet_ntop, inet_pton, AF_INET

from bson import Binary

from ming import Session, schema
from ming.datastore import create_datastore
from ming.odm import ThreadLocalODMSession, FieldProperty, Mapper
from ming.odm.declarative import MappedClass

bind = create_datastore('mongodb://localhost:27017/test')
doc_session = Session(bind)
session = ThreadLocalODMSession(doc_session=doc_session)

class CustomSchemaItem(schema.ParticularScalar):
    """An abstract base class for custom Ming types.
    
    Create a subclass with a class property called ``type``.
    It should be set to a callable that constructs an instance of
    a subclass of a type that can be stored in MongoDB either from
    the storage representation or as whatever the application uses
    to represent the data.

    """
    def _validate(self, value, **kw):
        if self._allow_none and value is None: return value
        try:
            value = self.type(value)
        except Exception:
            raise schema.Invalid("%s is not a %s" % (value, self.type), value, None)
        return value

class IPv4Address(Binary):
    def __new__(cls, data):
        if isinstance(data, Binary):
            # This case covers both "load from database" and "copy constructor"
            self = super(IPv4Address, cls).__new__(cls, data, data.subtype)
        else:
            # This constructs from the application data
            self = super(IPv4Address, cls).__new__(cls, inet_pton(AF_INET, data))
        self.__value = inet_ntop(AF_INET, self)
        return self
    
    @property
    def v(self):
        return self.__value
    
    def __repr__(self):
        return 'IPv4Address({0!r})'.format(self.__value)

class IPv4SchemaItem(CustomSchemaItem):
    type = IPv4Address
    
class ConnectionLog(MappedClass):
    class __mongometa__:
        name = 'connlog'
        session = session
    
    _id = FieldProperty(schema.ObjectId)
    ip = FieldProperty(IPv4SchemaItem)
    time = FieldProperty(datetime, if_missing=datetime.utcnow)

Mapper.compile_all()

if __name__ == '__main__':
    d = ConnectionLog(ip='192.0.2.1')
    print "Storing", d
    session.flush()
    session.clear()
    d = ConnectionLog.query.get()
    print "Loaded", d
    print "d.ip.v =", d.ip.v