python-zsi 2.1~a1-4 / usr / lib / python2.7 / dist-packages / ZSI / schema.py

# $Header$
'''XML Schema support
'''

from ZSI import _copyright, _seqtypes, _find_type, _get_element_nsuri_name, EvaluateException
from ZSI.wstools.Namespaces import SCHEMA, SOAP
from ZSI.wstools.Utility import SplitQName


def _get_type_definition(namespaceURI, name, **kw):
    return SchemaInstanceType.getTypeDefinition(namespaceURI, name, **kw)

def _get_global_element_declaration(namespaceURI, name, **kw):
    return SchemaInstanceType.getElementDeclaration(namespaceURI, name, **kw)

def _get_substitute_element(head, elt, ps):
    '''if elt matches a member of the head substitutionGroup, return 
    the GED typecode.

    head -- ElementDeclaration typecode, 
    elt -- the DOM element being parsed
    ps -- ParsedSoap Instance
    '''
    if not isinstance(head, ElementDeclaration):
        return None

    return ElementDeclaration.getSubstitutionElement(head, elt, ps)

def _has_type_definition(namespaceURI, name):
    return SchemaInstanceType.getTypeDefinition(namespaceURI, name) is not None

def _is_substitute_element(head, sub):
    '''if head and sub are both GEDs, and sub declares 
    head as its substitutionGroup then return True.

    head -- Typecode instance
    sub  -- Typecode instance
    '''
    if not isinstance(head, ElementDeclaration) or not isinstance(sub, ElementDeclaration):
        return False

    try:
        group = sub.substitutionGroup 
    except (AttributeError, TypeError):
        return False

    ged = GED(*group)

    # TODO: better way of representing element references.  Wrap them with
    # facets, and dereference when needed and delegate to..
    print (head.nspname == ged.nspname and head.pname == ged.pname)
    if head is ged or not (head.nspname == ged.nspname and head.pname == ged.pname):
        return False

    return True

#
# functions for retrieving schema items from 
# the global schema instance.
#
GED = _get_global_element_declaration
GTD = _get_type_definition


def WrapImmutable(pyobj, what):
    '''Wrap immutable instance so a typecode can be
    set, making it self-describing ie. serializable.
    '''
    return _GetPyobjWrapper.WrapImmutable(pyobj, what)

def RegisterBuiltin(arg):
    '''Add a builtin to be registered, and register it
    with the Any typecode.
    '''
    _GetPyobjWrapper.RegisterBuiltin(arg)
    _GetPyobjWrapper.RegisterAnyElement()

def RegisterAnyElement():
    '''register all Wrapper classes with the Any typecode.
    This allows instances returned by Any to be self-describing.
    ie. serializable.  AnyElement falls back on Any to parse
    anything it doesn't understand.
    '''
    return _GetPyobjWrapper.RegisterAnyElement()


class SchemaInstanceType(type):
    '''Register all types/elements, when hit already defined 
    class dont create a new one just give back reference.  Thus 
    import order determines which class is loaded.

    class variables:
        types -- dict of typecode classes definitions 
            representing global type definitions.
        elements -- dict of typecode classes representing 
            global element declarations.
        element_typecode_cache -- dict of typecode instances 
            representing global element declarations.
    '''
    types = {}
    elements = {}
    element_typecode_cache = {}
    #substitution_registry = {}
    
    def __new__(cls,classname,bases,classdict):
        '''If classdict has literal and schema register it as a
        element declaration, else if has type and schema register
        it as a type definition.
        '''
        if classname in ['ElementDeclaration', 'TypeDefinition', 'LocalElementDeclaration',]:
            return type.__new__(cls,classname,bases,classdict)

        if ElementDeclaration in bases:
            if classdict.has_key('schema') is False  or classdict.has_key('literal') is False: 
                raise AttributeError, 'ElementDeclaration must define schema and literal attributes'

            key = (classdict['schema'],classdict['literal'])
            if SchemaInstanceType.elements.has_key(key):
                return SchemaInstanceType.elements[key]

            # create global element declaration
            ged = SchemaInstanceType.elements[key] = type.__new__(cls,classname,bases,classdict)

            # TODO: Maybe I want access to all registrants??
            # 
            #if classdict.has_key('substitutionGroup'):
            #    sub = classdict.has_key('substitutionGroup')
            #    if not SchemaInstanceType.substitution_registry.has_key(sub):
            #        SchemaInstanceType.substitution_registry[sub] = [ged]
            #    else:
            #        SchemaInstanceType.substitution_registry[sub].append(ged)

            return ged

        if TypeDefinition in bases:
            if classdict.has_key('type') is None:
                raise AttributeError, 'TypeDefinition must define type attribute'

            key = classdict['type']
            if SchemaInstanceType.types.has_key(key) is False:
                SchemaInstanceType.types[key] = type.__new__(cls,classname,bases,classdict)
            return SchemaInstanceType.types[key]

        if LocalElementDeclaration in bases:
                return type.__new__(cls,classname,bases,classdict)

        raise TypeError, 'SchemaInstanceType must be an ElementDeclaration or TypeDefinition '

    def getTypeDefinition(cls, namespaceURI, name, lazy=False):
        '''Grab a type definition, returns a typecode class definition
        because the facets (name, minOccurs, maxOccurs) must be provided.
 
        Parameters:
           namespaceURI -- 
           name -- 
        '''
        klass = cls.types.get((namespaceURI, name), None)
        if lazy and klass is not None:
            return _Mirage(klass)
        return klass
    getTypeDefinition = classmethod(getTypeDefinition)

    def getElementDeclaration(cls, namespaceURI, name, isref=False, lazy=False):
        '''Grab an element declaration, returns a typecode instance
        representation or a typecode class definition.  An element 
        reference has its own facets, and is local so it will not be
        cached.

        Parameters:
            namespaceURI -- 
            name -- 
            isref -- if element reference, return class definition.
        '''
        key = (namespaceURI, name)
        if isref:
            klass = cls.elements.get(key,None)
            if klass is not None and lazy is True:
                return _Mirage(klass)
            return klass
 
        typecode = cls.element_typecode_cache.get(key, None)
        if typecode is None:
            tcls = cls.elements.get(key,None)
            if tcls is not None:
                typecode = cls.element_typecode_cache[key] = tcls()
                typecode.typed = False
            
        return typecode
    getElementDeclaration = classmethod(getElementDeclaration)


class ElementDeclaration:
    '''Typecodes subclass to represent a Global Element Declaration by
    setting class variables schema and literal.

    schema = namespaceURI
    literal = NCName
    substitutionGroup -- GED reference of form, (namespaceURI,NCName)
    '''
    __metaclass__ = SchemaInstanceType

    def checkSubstitute(self, typecode):
        '''If this is True, allow typecode to be substituted
        for "self" typecode.
        '''
        if not isinstance(typecode, ElementDeclaration): 
            return False

        try:
            nsuri,ncname = typecode.substitutionGroup
        except (AttributeError, TypeError):
            return False

        if (nsuri,ncname) != (self.schema,self.literal):
            # allow slop with the empty namespace 
            if not nsuri and not self.schema and ncname == self.literal:
                 return True

            return False

        sub = GED(self.schema, self.literal)
        if sub is None or sub is not typecode:
            return False

        return True

    def getSubstitutionElement(self, elt, ps):
        '''if elt matches a member of the head substitutionGroup, return 
        the GED typecode representation of the member.

        head -- ElementDeclaration typecode, 
        elt -- the DOM element being parsed
        ps -- ParsedSoap instance
        '''
        nsuri,ncname = _get_element_nsuri_name(elt)
        typecode = GED(nsuri,ncname)
        if typecode is None:
            return

        try:
            nsuri,ncname = typecode.substitutionGroup
        except (AttributeError, TypeError):
            return

        if (ncname == self.pname) and (nsuri == self.nspname or 
           (not nsuri and not self.nspname)):
             return typecode
       
        return 
 
 
class LocalElementDeclaration:
    '''Typecodes subclass to represent a Local Element Declaration.
    '''
    __metaclass__ = SchemaInstanceType
    

class TypeDefinition:
    '''Typecodes subclass to represent a Global Type Definition by
    setting class variable type.

    type = (namespaceURI, NCName)
    '''
    __metaclass__ = SchemaInstanceType
    
    def getSubstituteType(self, elt, ps):
        '''if xsi:type does not match the instance type attr,
        check to see if it is a derived type substitution.
        
        DONT Return the element's type.
        
        Parameters:
            elt -- the DOM element being parsed
            ps -- the ParsedSoap object.
        '''
        pyclass = SchemaInstanceType.getTypeDefinition(*self.type)
        if pyclass is None:
            raise EvaluateException(
                    'No Type registed for xsi:type=(%s, %s)' %
                    (self.type[0], self.type[1]), ps.Backtrace(elt))
            
        typeName = _find_type(elt)
        prefix,typeName = SplitQName(typeName)
        uri = ps.GetElementNSdict(elt).get(prefix)
        subclass = SchemaInstanceType.getTypeDefinition(uri, typeName)
        if subclass is None:
            raise EvaluateException(
                    'No registered xsi:type=(%s, %s), substitute for xsi:type=(%s, %s)' %
                    (uri, typeName, self.type[0], self.type[1]), ps.Backtrace(elt))
                    
        if not issubclass(subclass, pyclass) and subclass(None) and not issubclass(subclass, pyclass):
            raise TypeError(
                    'Substitute Type (%s, %s) is not derived from %s' %
                    (self.type[0], self.type[1], pyclass), ps.Backtrace(elt))

        return subclass((self.nspname, self.pname))
    
    

class _Mirage:
    '''Used with SchemaInstanceType for lazy evaluation, eval during serialize or 
    parse as needed.  Mirage is callable, TypeCodes are not.  When called it returns the
    typecode.  Tightly coupled with generated code.
    
    NOTE: **Must Use ClassType** for intended MRO of __call__ since setting it in
    an instance attribute rather than a class attribute (will not work for object).
    '''
    def __init__(self, klass):
        self.klass = klass
        self.__reveal = False
        self.__cache = None
        if issubclass(klass, ElementDeclaration):
            self.__call__ = self._hide_element
            
    def __str__(self):
        msg = "<Mirage id=%s, Local Element %s>"
        if issubclass(self.klass, ElementDeclaration):
            msg = "<Mirage id=%s, GED %s>"
        return  msg %(id(self), self.klass)
        
    def _hide_type(self, pname, aname, minOccurs=0, maxOccurs=1, nillable=False, 
                   **kw):
        self.__call__ = self._reveal_type
        self.__reveal = True
        
        # store all attributes, make some visable for pyclass_type
        self.__kw = kw
        self.minOccurs,self.maxOccurs,self.nillable = minOccurs,maxOccurs,nillable
        self.nspname,self.pname,self.aname = None,pname,aname
        if type(self.pname) in (tuple,list):
            self.nspname,self.pname = pname
        
        return self
        
    def _hide_element(self, minOccurs=0, maxOccurs=1, nillable=False, **kw):
        self.__call__ = self._reveal_element
        self.__reveal = True
        
        # store all attributes, make some visable for pyclass_type
        self.__kw = kw
        self.nspname = self.klass.schema
        self.pname = self.klass.literal
        #TODO: Fix hack
        #self.aname = '_%s' %self.pname
        self.minOccurs,self.maxOccurs,self.nillable = minOccurs,maxOccurs,nillable
        
        return self
    
    def _reveal_type(self):
        if self.__cache is None:
            self.__cache = self.klass(pname=self.pname, 
                            aname=self.aname, minOccurs=self.minOccurs, 
                            maxOccurs=self.maxOccurs, nillable=self.nillable, 
                            **self.__kw)
        return self.__cache
        
    def _reveal_element(self):
        if self.__cache is None:
            self.__cache = self.klass(minOccurs=self.minOccurs, 
                            maxOccurs=self.maxOccurs, nillable=self.nillable, 
                            **self.__kw)
        return self.__cache
    
    __call__ = _hide_type


class _GetPyobjWrapper:
    '''Get a python object that wraps data and typecode.  Used by
    <any> parse routine, so that typecode information discovered
    during parsing is retained in the pyobj representation
    and thus can be serialized.
    '''
    types_dict = dict()

    def RegisterBuiltin(cls, arg):
        '''register a builtin, create a new wrapper.
        '''
        if arg in cls.types_dict:
            raise RuntimeError, '%s already registered' %arg
        class _Wrapper(arg):
            'Wrapper for builtin %s\n%s' %(arg, cls.__doc__)
        _Wrapper.__name__ = '_%sWrapper' %arg.__name__
        cls.types_dict[arg] = _Wrapper
    RegisterBuiltin = classmethod(RegisterBuiltin)
        
    def RegisterAnyElement(cls):
        '''If find registered TypeCode instance, add Wrapper class 
        to TypeCode class serialmap and Re-RegisterType.  Provides
        Any serialzation of any instances of the Wrapper.
        '''
        for k,v in cls.types_dict.items():
            what = Any.serialmap.get(k)
            if what is None: continue
            if v in what.__class__.seriallist: continue
            what.__class__.seriallist.append(v)
            RegisterType(what.__class__, clobber=1, **what.__dict__)
    RegisterAnyElement = classmethod(RegisterAnyElement)

    def WrapImmutable(cls, pyobj, what):
        '''return a wrapper for pyobj, with typecode attribute set.
        Parameters:
            pyobj -- instance of builtin type (immutable)
            what -- typecode describing the data
        '''
        d = cls.types_dict
        if type(pyobj) is bool:  
            pyclass = d[int]
        elif d.has_key(type(pyobj)) is True:
            pyclass = d[type(pyobj)]
        else:
            raise TypeError,\
               'Expecting a built-in type in %s (got %s).' %(
                d.keys(),type(pyobj))

        newobj = pyclass(pyobj)
        newobj.typecode = what
        return newobj
    WrapImmutable = classmethod(WrapImmutable)
    

from TC import Any, RegisterType

if __name__ == '__main__': print _copyright