[Zope3-Users] newbie design questions for UI to external data

Shaun Cutts shaun at cuttshome.net
Mon Feb 13 22:33:17 EST 2006


Um...

I guess I must have asked too many questions at once :) I've implemented
a first pass of a container for external data and got it working on some
of my collections.

I've included the (first pass of the) base DBContainer and DBContained
objects for criticism, and would be very grateful if anyone would like
to take a crack at it.

Keep in mind, this is my first time working with the system, so I'm sure
I'm going about some things the wrong way. In particular, I already
think:

1) Instead of have the contained object update the container, I should
rely on the event notification system.

2) Right now, the system works by translating objects "at the border"
(in IExternalContainer). Some translation is necessary, for instance, to
move from mx.DateTime to datetime.datetime, but still I think I should
somehow be making use of the "adaptor" interface. 

3) Along these same lines, IDBContainer._containedType should really be
an interface (Object( schema = IDBContained ))

Note: is there a tutorial on writing containers anywhere I should have
read? I mainly figured this out by banging on it and fishing around in
the code. I'd love to figure out, for instance, what is really happening
with the traversals (with some interaction diagrams). I do think it was
harder than it should have been. (But, then again, I think that about
most things...:))

BTW in my humble opinion, ILocation.__name__ is not well named. When I
first got an error referring to __name__ I thought it was expecting a
class object. And what happens when, for some strange reason, someone
wants to put a class in a container, and doesn't like its default name?

Thanks,
- Shaun
------------------------------------------------------------------------
-------------

class IExternalContainer( Interface ):
    """an external container interface."""

    def add( obj ):
        """add 'obj'"""

    def update( obj ):
        """update obj"""

    def delete( obj ):
        """delete obj"""

    def connect( obj ):
        """establish a connection to external database"""

    def __iter__( ):
        """iterates through external objects"""

class IDBContainer( IContainer, ILocation ):
    """container that lives in zope, but contains objects
    made persistent by an external container."""

    _database = Object( schema = IExternalContainer )        
    _containedType = Attribute( "type of object contained" )

    def __setitem__( key, obj ): 
        """Add a IDBContained object.""" 

    def _updateItem( key, newKey = None ):
        """notify that object has been modified.
        
        If the key has been changed, 'key' should be the old
        key of the object, and 'newKey' should be set to the
        new key."""

    def __delitem__( key ):
        """delete item, removing it both from this view
        and from external storage."""

    def getAttributeNames():
        """get names of attributes of contained object"""

class IDBContained( IContained, ILocation ):
    """Contained in IDBContainer. Must notify its parent
    if it is changed. Because of this, it must also know how
    to get its key.
    """
    __parent__ = Field(
        constraint = ContainerTypesConstraint( IDBContainer ) )

    def getZopeKey():
        """return ascii str or unicode key."""

IDBContainer[ '__setitem__' ].precondition = ItemTypePrecondition(
IDBContained )
class DBContainer( IterableUserDict, Contained ):
    """
    Implements L{IDBContainer}

    >>> from zope.interface.verify import verifyClass
    >>> verifyClass( IDBContainer, DBContainer )
    True

    """
    implements( IDBContainer )
    
    # prevent browser from choosing name
    nameAllowed = False
    
    # defaults for IDBContainer
    _database = None
    _containedType = None
    
    # defaults for ILocation (base of IDBContainer)
    __parent__ = None
    __name__ = None
    
    def __init__( self  ):
        super( DBContainer, self ).__init__( )
        self.__setstate__()

    def _filterKey( self, newKey ):
        """check that key is ascii string or unicode, and convert
        to unicode."""
        
        if newKey is None:
            TypeError( "key can't be None" )
        elif isinstance( newKey, str ):
            try:
                newKey = unicode( newKey )
            except UnicodeError:
                raise TypeError("name not unicode or ascii string")

        elif not isinstance( newKey, unicode ):
            raise TypeError( "key '%s' must be ascii or unicode string"
% repr( newKey ) )

        return newKey

    def __setitem__( self, key, obj ):
        "Implements
L{cranedata.web.interfaces.IFundContainer.__setitem__}."
        key = self._filterKey( key )
        if not isinstance( obj, self._containedType ):
            raise TypeError( "object '%s' must be a %s" % ( repr( obj ),
self._containedType.__name__ ) )

        self._database.add( obj )
        setitem( self, self.data.__setitem__, key, obj )

    def __getitem__( self, key ):
        return self.data[ self._filterKey( key ) ]

    def __delitem__( self, key ):
        obj = self.data.pop( self._filterKey( key ) )
        self._database.delete( obj )
        uncontained( obj, self, obj.__name__ )

    def _updateItem( self, oldKey, newKey = None ):
        """implements L{IFundContainer.__setitem__}
        be specified"""
        oldKey = self._filterKey( oldKey )
        if newKey is not None:
            obj = self.data.pop( oldKey )
        else:
            obj = self.data[ oldKey ]
            
        self._database.update( obj )

        if newKey is not None:
            self.data[ self._filterKey( newKey ) ] = obj


    # need __getstate__, __setstate__ for ZODB persistence
    
    def __getstate__( self ):
        # we have to remember our tree-location
        return {
            '__name__' : self.__name__,
            '__parent__':self.__parent__,
            #'__annotations__': self.__annotations__
            }

    def __setstate__( self, dct = None ):
        if dct:
            self.__parent__ = dct[ '__parent__' ]
            self.__name__ = dct[ '__name__' ]
            self.data = {}
            
        self._database.connect()
        self._load()
        
    def _load( self ):
        for row in self._database:
            self.data[ row.getZopeKey() ] = row
            row.__parent__ = self

    def getAttributeNames( self ):
        return self._containedType.getAttributeNames()


class DBContained( Location ):
    """Object meant to be contained by DBContained.

    It knows own key, and notifies parent on update to any attribute but
    a 'magic' one (start and end w/ `__`). 

    @cvar __keys__: attributes in keys. If not set, key is all
attributes
        except magic ones
    @cvar __rowInterface__: an Interface that defines the attributes of
the object.
        If this is defined, it is used to validate data, and reject
attributes
        not in list.

    @warning: no action is taken on __del__

    >>> r = DBContained( a = 1, b = 2, c = 3 )
    >>> l = r.__dict__.items()
    >>> l.sort()
    >>> print l
    [('a', 1), ('b', 2), ('c', 3)]
    >>> r.getZopeKey()
    u'1_2_3'
    >>> s = DBContained( a = 1, b = 'spring', c = 'fall', __keys__ =
['c', 'a'] )
    >>> s.getZopeKey()
    u'fall_1'
    >>> class Parent:
    ...     def _updateItem( self, oldKey, newKey ):
    ...         print repr( oldKey )
    ...         print repr( newKey )
    ...
    >>> r.__parent__ = Parent()
    >>> r.a = 4
    u'1_2_3'
    u'4_2_3'
    >>> r.bb = 'winter'
    u'4_2_3'
    u'4_2_winter_3'
    >>> r.__x__ = 1

    >>> from zope.interface import Interface
    >>> from zope.schema import Int
    >>> class IR( Interface ):
    ...     a = Int( )
    ...     b = Int( )
    >>> class RR( DBContained ):
    ...     __rowInterface__ = IR
    >>> r = RR( a = 1, c = 2 )
    Traceback (most recent call last):
        ...
    TypeError: attr c not in row interface <InterfaceClass
cranedata.web.container.IR>
    >>> r = RR( a = 1, b = 2 )
    >>> RR.getAttributeNames()
    ('a', 'b')
    >>> r = RR( a = 'foo', b = 2 )
    Traceback (most recent call last):
    ...
    WrongType: ('foo', (<type 'int'>, <type 'long'>))

    We convert strings to unicode:
    
    >>> class IS( Interface ):
    ...     s = Text( )
    >>> class SS( DBContained ):
    ...     __rowInterface__ = IS
    >>> r = SS( s = 'hello' )
    >>> r.s
    u'hello'
    
    """

    implements( IDBContained )

    def __init__( self, **kw ):
        self.__dict__.update( kw )
        try:
            iface = self.__rowInterface__
        except AttributeError:
            pass
        else:
            fields = dict( iface.namesAndDescriptions( all = True ) )
            for attr, val in self.__dict__.iteritems():
                if self.isMagicAttribute( attr ):
                    continue
                field = fields.pop( attr )
                if field is None:
                    raise TypeError( "attr %s not in row interface %s" %
\
                                     ( attr, str( iface ) ) )
                
                # convert strings to unicode if required
                if isinstance( field, Text ) and isinstance( val, str ):
                    setattr( self, attr, unicode( val ) )
                # truncate datetime to date if required
                if isinstance( field, date ) and isinstance( val,
datetime ):
                    setattr(
                        self, attr, date( date.fromordinal(
val.toordinal() ) ) )
                               
                bound = field.bind( self )
                bound.validate( bound.get( self ) )

            # set defaults, or raise error when required and no default
            for fname, field in fields.iteritems():
                if not IMethod.providedBy( field ):
                    if IField.providedBy( field ):
                        if field.required:
                            if field.default:
                                setattr( self, fname, field.default )
                            else:
                                raise RequiredMissing( "missing: %s" %
fname )

                        else:
                            if field.missing_value:
                                setattr( self, fname,
field.missing_value )
                            else:
                                setattr( self, fname, None )
                    else:
                        setattr( self, fname, None )
                    
                
        super( DBContained, self ).__init__( ) 


    @staticmethod
    def isMagicAttribute( attr ):
        return attr.startswith( '__' ) and attr.endswith( '__' )

    def getZopeKeyAttributes( self ):
        try:
            keys = self.__keys__
        except AttributeError:
            keys = filter(
                lambda k: not self.isMagicAttribute( k ),
self.__dict__.keys() )
            keys.sort()
    
            
        return keys

    @classmethod
    def getAttributeNames( self ):
        return tuple( self.__rowInterface__.names( all = True ) )
        

    def getAttributeDictionary( self ):
        """return attr:val dictionary for 'non-magic' attributes.

        >>> r = DBContained( a = 1, b = 2, __keys__ = ( 'a', ), __foo__
= 23 )
        >>> sorted( r.getAttributeDictionary().items() )
        [('a', 1), ('b', 2)]

        """
        return dict( ifilter(
            lambda item: not self.isMagicAttribute( item[ 0 ] ),
            self.__dict__.iteritems() ) )

    def getZopeKey( self ):
        keys = self.getZopeKeyAttributes()
        key = [ str( getattr( self, col, u'None' ) ) for col in
self.getZopeKeyAttributes() ]
        if len( key ) > 0:
            return u'_'.join( key )
        else:
            return u'*'

    def __setattr__( self, attr, val ):
        oldKey = self.getZopeKey()
        self.__dict__[ attr ] = val
        # notify parent of change, but not for 'magic names':
        if ( not attr.startswith( '__' ) or not attr.endswith( '__' ) )
and \
               self.__parent__ != None:
            self.__parent__._updateItem( oldKey, self.getZopeKey() )


    def checkZopeKey( self, key ):
        assert key == self.getZopeKey()
        
    __name__ = property( getZopeKey, checkZopeKey )
            




More information about the Zope3-users mailing list