def getPort(self):
"""
Returns a Port object of the type CF__POA.Port.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
PortClass = classobj('PortClass',
(CF__POA.Port,),
{'connectPort':self.connectPort,
'disconnectPort':self.disconnectPort})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
python类classobj()的实例源码
def getPort(self):
"""
Returns a Port object of the type CF__POA.Port.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
PortClass = classobj('PortClass',
(CF__POA.Port,),
{'connectPort':self.connectPort,
'disconnectPort':self.disconnectPort})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def getPort(self):
"""
Returns a Port object of the type CF__POA.Port.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
PortClass = classobj('PortClass',
(CF__POA.Port,),
{'connectPort':self.connectPort,
'disconnectPort':self.disconnectPort})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def makeSQLTests(base, suffix, globals):
"""Make a test case for every db connector which can connect.
@param base: Base class for test case. Additional base classes
will be a DBConnector subclass and unittest.TestCase
@param suffix: A suffix used to create test case names. Prefixes
are defined in the DBConnector subclasses.
"""
connectors = [GadflyConnector, SQLiteConnector, PyPgSQLConnector,
PsycopgConnector, MySQLConnector, FirebirdConnector]
for connclass in connectors:
name = connclass.TEST_PREFIX + suffix
import new
klass = new.classobj(name, (connclass, base, unittest.TestCase), {})
globals[name] = klass
# GadflyADBAPITestCase SQLiteADBAPITestCase PyPgSQLADBAPITestCase
# PsycopgADBAPITestCase MySQLADBAPITestCase FirebirdADBAPITestCase
def makeSQLTests(base, suffix, globals):
"""Make a test case for every db connector which can connect.
@param base: Base class for test case. Additional base classes
will be a DBConnector subclass and unittest.TestCase
@param suffix: A suffix used to create test case names. Prefixes
are defined in the DBConnector subclasses.
"""
connectors = [GadflyConnector, SQLiteConnector, PyPgSQLConnector,
PsycopgConnector, MySQLConnector, FirebirdConnector]
for connclass in connectors:
name = connclass.TEST_PREFIX + suffix
import new
klass = new.classobj(name, (connclass, base, unittest.TestCase), {})
globals[name] = klass
# GadflyADBAPITestCase SQLiteADBAPITestCase PyPgSQLADBAPITestCase
# PsycopgADBAPITestCase MySQLADBAPITestCase FirebirdADBAPITestCase
def factoryclass (base, factory):
def init (self, *args, **kwargs):
base.__init__(self, *args, **kwargs)
factory.__init__(self)
return new.classobj(base.__name__, (base, factory), { '__init__': init })
def getPort(self):
"""
Returns a Port object of the same type as the one specified as the
porttype argument during the object instantiation. It uses the
classobj from the new module to generate a class on runtime.
The classobj generates a class using the following arguments:
name: The name of the class to generate
bases: A tuple containing all the base classes to use
dct: A dictionary containing all the attributes such as
functions, and class variables
It is important to notice that the porttype is a BULKIO__POA type and
not a BULKIO type. The reason is because it is used to generate a
Port class that will be returned when the getPort() is invoked. The
returned class is the one acting as a server and therefore must be a
Portable Object Adapter rather and a simple BULKIO object.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
PortClass = classobj('PortClass',
(self.port_type,),
{'pushPacket':self.pushPacket,
'pushSRI':self.pushSRI})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def getPort(self):
"""
Returns a Port object of the same type as the one specified as the
porttype argument during the object instantiation. It uses the
classobj from the new module to generate a class on runtime.
The classobj generates a class using the following arguments:
name: The name of the class to generate
bases: A tuple containing all the base classes to use
dct: A dictionary containing all the attributes such as
functions, and class variables
It is important to notice that the porttype is a BULKIO__POA type and
not a BULKIO type. The reason is because it is used to generate a
Port class that will be returned when the getPort() is invoked. The
returned class is the one acting as a server and therefore must be a
Portable Object Adapter rather and a simple BULKIO object.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
if self.port_type == BULKIO__POA.dataXML:
pushPacket = self.pushPacketXML
else:
pushPacket = self.pushPacket
PortClass = classobj('PortClass',
(self.port_type,),
{'pushPacket':pushPacket,
'pushSRI':self.pushSRI})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def getPort(self):
PortClass = classobj('PortClass',
(BULKIO__POA.dataSDDS,),
{'_get_attachmentIds':self._get_attachmentIds,
'_get_attachedStreams':self._get_attachedStreams,
'_get_usageState':self._get_usageState,
'getUser':self.getUser,
'getStreamDefinition':self.getStreamDefinition,
'pushSRI':self.pushSRI,
'attach':self.attach,
'detach':self.detach})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def getPort(self):
"""
Returns a Port object of the same type as the one specified as the
porttype argument during the object instantiation. It uses the
classobj from the new module to generate a class on runtime.
The classobj generates a class using the following arguments:
name: The name of the class to generate
bases: A tuple containing all the base classes to use
dct: A dictionary containing all the attributes such as
functions, and class variables
It is important to notice that the porttype is a BULKIO__POA type and
not a BULKIO type. The reason is because it is used to generate a
Port class that will be returned when the getPort() is invoked. The
returned class is the one acting as a server and therefore must be a
Portable Object Adapter rather and a simple BULKIO object.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
if self.port_type == BULKIO__POA.dataXML:
pushPacket = self.pushPacketXML
else:
pushPacket = self.pushPacket
PortClass = classobj('PortClass',
(self.port_type,),
{'pushPacket':pushPacket,
'pushSRI':self.pushSRI})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def getPort(self):
"""
Returns a Port object of the same type as the one specified as the
porttype argument during the object instantiation. It uses the
classobj from the new module to generate a class on runtime.
The classobj generates a class using the following arguments:
name: The name of the class to generate
bases: A tuple containing all the base classes to use
dct: A dictionary containing all the attributes such as
functions, and class variables
It is important to notice that the porttype is a BULKIO__POA type and
not a BULKIO type. The reason is because it is used to generate a
Port class that will be returned when the getPort() is invoked. The
returned class is the one acting as a server and therefore must be a
Portable Object Adapter rather and a simple BULKIO object.
"""
# The classobj generates a class using the following arguments:
#
# name: The name of the class to generate
# bases: A tuple containing all the base classes to use
# dct: A dictionary containing all the attributes such as
# functions, and class variables
PortClass = classobj('PortClass',
(self.port_type,),
{'pushPacket':self.pushPacket,
'pushSRI':self.pushSRI})
# Create a port using the generate Metaclass and return an instance
port = PortClass()
return port._this()
def save_classobj(self, obj):
""" Save an interactively defined classic class object by value """
if obj.__module__ == '__main__':
args = (obj.__name__, obj.__bases__, obj.__dict__)
self.save_reduce(new.classobj, args, obj=obj)
else:
pickle.Pickler.save_global(self, obj, name)
def test_sf_payload(self):
with self.assertRaises(spickle.UnpicklingError):
def nasty(module, function, *args):
return pickle.dumps(new.classobj(function, (), {
'__getinitargs__': lambda self, arg=args: arg,
'__module__': module
})())
t = nasty("subprocess", "Popen", ("/bin/ls", "/tmp"))
spickle.loads(t)
def hook__setattr__(obj):
if not hasattr(obj,'__attrproxy__'):
C = obj.__class__
import new
obj.__class__=new.classobj(C.__name__,(C,)+C.__bases__,
{'__attrproxy__':[],
'__setattr__':lambda self,k,v,osa=getattr(obj,'__setattr__',None),hook=hook: hook(self,k,v,osa)})
def watchObject(self, object, identifier, callback):
"""Watch the given object.
Whenever I think the object might have changed, I'll send an
ObjectLink of it to the callback.
The identifier argument is used to generate identifiers for
objects which are members of this one.
"""
if type(object) is not types.InstanceType:
raise TypeError, "Sorry, can only place a watch on Instances."
# uninstallers = []
dct = {}
reflect.addMethodNamesToDict(object.__class__, dct, '')
for k in object.__dict__.keys():
dct[k] = 1
members = dct.keys()
clazzNS = {}
clazz = new.classobj('Watching%s%X' %
(object.__class__.__name__, id(object)),
(_MonkeysSetattrMixin, object.__class__,),
clazzNS)
clazzNS['_watchEmitChanged'] = new.instancemethod(
lambda slf, i=identifier, b=self, cb=callback:
cb(b.browseObject(slf, i)),
None, clazz)
# orig_class = object.__class__
object.__class__ = clazz
for name in members:
m = getattr(object, name)
# Only hook bound methods.
if ((type(m) is types.MethodType)
and (m.im_self is not None)):
# What's the use of putting watch monkeys on methods
# in addition to __setattr__? Well, um, uh, if the
# methods modify their attributes (i.e. add a key to
# a dictionary) instead of [re]setting them, then
# we wouldn't know about it unless we did this.
# (Is that convincing?)
monkey = _WatchMonkey(object)
monkey.install(name)
# uninstallers.append(monkey.uninstall)
# XXX: This probably prevents these objects from ever having a
# zero refcount. Leak, Leak!
## self.watchUninstallers[object] = uninstallers
def create_table(self,table,migrate=True,fake_migrate=False, polymodel=None):
myfields = {}
for field in table:
if isinstance(polymodel,Table) and field.name in polymodel.fields():
continue
attr = {}
if isinstance(field.custom_qualifier, dict):
#this is custom properties to add to the GAE field declartion
attr = field.custom_qualifier
field_type = field.type
if isinstance(field_type, SQLCustomType):
ftype = self.types[field_type.native or field_type.type](**attr)
elif isinstance(field_type, gae.Property):
ftype = field_type
elif field_type.startswith('id'):
continue
elif field_type.startswith('decimal'):
precision, scale = field_type[7:].strip('()').split(',')
precision = int(precision)
scale = int(scale)
ftype = GAEDecimalProperty(precision, scale, **attr)
elif field_type.startswith('reference'):
if field.notnull:
attr = dict(required=True)
referenced = field_type[10:].strip()
ftype = self.types[field_type[:9]](referenced, **attr)
elif field_type.startswith('list:reference'):
if field.notnull:
attr['required'] = True
referenced = field_type[15:].strip()
ftype = self.types[field_type[:14]](**attr)
elif field_type.startswith('list:'):
ftype = self.types[field_type](**attr)
elif not field_type in self.types\
or not self.types[field_type]:
raise SyntaxError('Field: unknown field type: %s' % field_type)
else:
ftype = self.types[field_type](**attr)
myfields[field.name] = ftype
if not polymodel:
table._tableobj = classobj(table._tablename, (gae.Model, ), myfields)
elif polymodel==True:
table._tableobj = classobj(table._tablename, (PolyModel, ), myfields)
elif isinstance(polymodel,Table):
table._tableobj = classobj(table._tablename, (polymodel._tableobj, ), myfields)
else:
raise SyntaxError("polymodel must be None, True, a table or a tablename")
return None
def watchObject(self, object, identifier, callback):
"""Watch the given object.
Whenever I think the object might have changed, I'll send an
ObjectLink of it to the callback.
The identifier argument is used to generate identifiers for
objects which are members of this one.
"""
if type(object) is not types.InstanceType:
raise TypeError, "Sorry, can only place a watch on Instances."
# uninstallers = []
dct = {}
reflect.addMethodNamesToDict(object.__class__, dct, '')
for k in object.__dict__.keys():
dct[k] = 1
members = dct.keys()
clazzNS = {}
clazz = new.classobj('Watching%s%X' %
(object.__class__.__name__, id(object)),
(_MonkeysSetattrMixin, object.__class__,),
clazzNS)
clazzNS['_watchEmitChanged'] = new.instancemethod(
lambda slf, i=identifier, b=self, cb=callback:
cb(b.browseObject(slf, i)),
None, clazz)
# orig_class = object.__class__
object.__class__ = clazz
for name in members:
m = getattr(object, name)
# Only hook bound methods.
if ((type(m) is types.MethodType)
and (m.im_self is not None)):
# What's the use of putting watch monkeys on methods
# in addition to __setattr__? Well, um, uh, if the
# methods modify their attributes (i.e. add a key to
# a dictionary) instead of [re]setting them, then
# we wouldn't know about it unless we did this.
# (Is that convincing?)
monkey = _WatchMonkey(object)
monkey.install(name)
# uninstallers.append(monkey.uninstall)
# XXX: This probably prevents these objects from ever having a
# zero refcount. Leak, Leak!
## self.watchUninstallers[object] = uninstallers
def create_table(self, table, migrate=True, fake_migrate=False, polymodel=None):
myfields = {}
for field in table:
if isinstance(polymodel, Table) and field.name in polymodel.fields():
continue
attr = {}
if isinstance(field.custom_qualifier, dict):
#this is custom properties to add to the GAE field declartion
attr = field.custom_qualifier
field_type = field.type
if isinstance(field_type, SQLCustomType):
ftype = self.types[field_type.native or field_type.type](**attr)
elif isinstance(field_type, ((self.use_ndb and ndb.Property) or gae.Property)):
ftype = field_type
elif field_type.startswith('id'):
continue
elif field_type.startswith('decimal'):
precision, scale = field_type[7:].strip('()').split(',')
precision = int(precision)
scale = int(scale)
dec_cls = (self.use_ndb and NDBDecimalProperty) or GAEDecimalProperty
ftype = dec_cls(precision, scale, **attr)
elif field_type.startswith('reference'):
if field.notnull:
attr = dict(required=True)
ftype = self.types[field_type[:9]](**attr)
elif field_type.startswith('list:reference'):
if field.notnull:
attr['required'] = True
ftype = self.types[field_type[:14]](**attr)
elif field_type.startswith('list:'):
ftype = self.types[field_type](**attr)
elif not field_type in self.types or not self.types[field_type]:
raise SyntaxError('Field: unknown field type: %s' % field_type)
else:
ftype = self.types[field_type](**attr)
myfields[field.name] = ftype
if not polymodel:
model_cls = (self.use_ndb and ndb.Model) or gae.Model
table._tableobj = classobj(table._tablename, (model_cls, ), myfields)
if self.use_ndb:
# Set NDB caching variables
if self.ndb_settings and (table._tablename in self.ndb_settings):
for k, v in self.ndb_settings.iteritems():
setattr(table._tableobj, k, v)
elif polymodel==True:
pm_cls = (self.use_ndb and NDBPolyModel) or PolyModel
table._tableobj = classobj(table._tablename, (pm_cls, ), myfields)
elif isinstance(polymodel, Table):
table._tableobj = classobj(table._tablename, (polymodel._tableobj, ), myfields)
else:
raise SyntaxError("polymodel must be None, True, a table or a tablename")
return None
