我们从Python开源项目中,提取了以下18个代码示例,用于说明如何使用new.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 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 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 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