本文整理汇总了Python中twisted.python.reflect.namedClass函数的典型用法代码示例。如果您正苦于以下问题:Python namedClass函数的具体用法?Python namedClass怎么用?Python namedClass使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了namedClass函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: startService
def startService(self):
"""
Start the service.
"""
directory = directoryFromConfig(self.config)
# Load proxy assignments from XML if specified
if self.config.ProxyLoadFromFile:
proxydbClass = namedClass(self.config.ProxyDBService.type)
calendaruserproxy.ProxyDBService = proxydbClass(
**self.config.ProxyDBService.params)
loader = XMLCalendarUserProxyLoader(self.config.ProxyLoadFromFile)
yield loader.updateProxyDB()
# Populate the group membership cache
if (self.config.GroupCaching.Enabled and
self.config.GroupCaching.EnableUpdater):
proxydb = calendaruserproxy.ProxyDBService
if proxydb is None:
proxydbClass = namedClass(self.config.ProxyDBService.type)
proxydb = proxydbClass(**self.config.ProxyDBService.params)
updater = GroupMembershipCacheUpdater(proxydb,
directory,
self.config.GroupCaching.UpdateSeconds,
self.config.GroupCaching.ExpireSeconds,
self.config.GroupCaching.LockSeconds,
namespace=self.config.GroupCaching.MemcachedPool,
useExternalProxies=self.config.GroupCaching.UseExternalProxies)
yield updater.updateCache(fast=True)
# Set in motion the work queue based updates:
yield scheduleNextGroupCachingUpdate(self.store, 0)
uid, gid = getCalendarServerIDs(self.config)
dbPath = os.path.join(self.config.DataRoot, "proxies.sqlite")
if os.path.exists(dbPath):
os.chown(dbPath, uid, gid)
# Process old inbox items
self.store.setMigrating(True)
yield self.processInboxItems()
self.store.setMigrating(False)
# Migrate mail tokens from sqlite to store
yield migrateTokensToStore(self.config.DataRoot, self.store)
# Set mail polling in motion
if self.config.Scheduling.iMIP.Enabled:
yield scheduleNextMailPoll(self.store, 0)
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:49,代码来源:upgrade.py
示例2: send_add_attachment
def send_add_attachment(self, objectResource, rids, content_type, filename, stream):
"""
Managed attachment addAttachment call.
@param objectResource: child resource having an attachment added
@type objectResource: L{CalendarObject}
@param rids: list of recurrence ids
@type rids: C{list}
@param content_type: content type of attachment data
@type content_type: L{MimeType}
@param filename: name of attachment
@type filename: C{str}
@param stream: attachment data stream
@type stream: L{IStream}
"""
actionName = "add-attachment"
shareeView = objectResource._parentCollection
action, recipient = self._send(actionName, shareeView, objectResource)
action["rids"] = rids
action["filename"] = filename
result = yield self.sendRequest(shareeView._txn, recipient, action, stream, content_type)
if result["result"] == "ok":
returnValue(result["value"])
elif result["result"] == "exception":
raise namedClass(result["class"])(result["message"])
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:26,代码来源:conduit.py
示例3: raise_error
def raise_error():
# failure.parents[-1] will be the exception class for local
# failures and the string name of the exception class
# for remote failures (which might not exist in our
# namespace)
#
# failure.value will be the tuple of arguments to the
# exception in the local case, or a string
# representation of that in the remote case (see
# pb.CopyableFailure.getStateToCopy()).
#
# we can only reproduce a remote exception if the
# exception class is in our namespace, and it only takes
# one string argument. if either condition is not true,
# we wrap the strings in a default Exception.
k, v = failure.parents[-1], failure.value
try:
if isinstance(k, str):
k = reflect.namedClass(k)
if isinstance(v, tuple):
e = k(*v)
else:
e = k(v)
except Exception:
e = Exception('%s: %r' % (failure.type, v))
raise e
开发者ID:ylatuya,项目名称:Flumotion,代码行数:26,代码来源:defer.py
示例4: send_update_attachment
def send_update_attachment(self, objectResource, managed_id, content_type, filename, stream):
"""
Managed attachment updateAttachment call.
@param objectResource: child resource having an attachment added
@type objectResource: L{CalendarObject}
@param managed_id: managed-id to update
@type managed_id: C{str}
@param content_type: content type of attachment data
@type content_type: L{MimeType}
@param filename: name of attachment
@type filename: C{str}
@param stream: attachment data stream
@type stream: L{IStream}
"""
actionName = "update-attachment"
shareeView = objectResource._parentCollection
action, recipient = self._send(actionName, shareeView, objectResource)
action["managedID"] = managed_id
action["filename"] = filename
result = yield self.sendRequest(shareeView._txn, recipient, action, stream, content_type)
if result["result"] == "ok":
returnValue(result["value"])
elif result["result"] == "exception":
raise namedClass(result["class"])(result["message"])
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:26,代码来源:conduit.py
示例5: _simple_send
def _simple_send(self, actionName, shareeView, objectResource=None, transform=None, args=None, kwargs=None):
"""
A simple send operation that returns a value.
@param actionName: name of the action.
@type actionName: C{str}
@param shareeView: sharee resource being operated on.
@type shareeView: L{CommonHomeChildExternal}
@param objectResource: the resource being operated on, or C{None} for classmethod.
@type objectResource: L{CommonObjectResourceExternal}
@param transform: a function used to convert the JSON result into return values.
@type transform: C{callable}
@param args: list of optional arguments.
@type args: C{list}
@param kwargs: optional keyword arguments.
@type kwargs: C{dict}
"""
action, recipient = self._send(actionName, shareeView, objectResource)
if args is not None:
action["arguments"] = args
if kwargs is not None:
action["keywords"] = kwargs
result = yield self.sendRequest(shareeView._txn, recipient, action)
if result["result"] == "ok":
returnValue(result["value"] if transform is None else transform(result["value"], shareeView, objectResource))
elif result["result"] == "exception":
raise namedClass(result["class"])(result["message"])
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:28,代码来源:conduit.py
示例6: send_freebusy
def send_freebusy(
self,
calresource,
timerange,
matchtotal,
excludeuid,
organizer,
organizerPrincipal,
same_calendar_user,
servertoserver,
event_details,
):
action, recipient = self._send("freebusy", calresource)
action["timerange"] = [timerange.start.getText(), timerange.end.getText()]
action["matchtotal"] = matchtotal
action["excludeuid"] = excludeuid
action["organizer"] = organizer
action["organizerPrincipal"] = organizerPrincipal
action["same_calendar_user"] = same_calendar_user
action["servertoserver"] = servertoserver
action["event_details"] = event_details
result = yield self.sendRequest(calresource._txn, recipient, action)
if result["result"] == "ok":
returnValue((result["fbresults"], result["matchtotal"],))
elif result["result"] == "exception":
raise namedClass(result["class"])(result["message"])
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:26,代码来源:conduit.py
示例7: test_namedClassLookup
def test_namedClassLookup(self):
"""
L{namedClass} should return the class object for the name it is passed.
"""
self.assertIs(
reflect.namedClass("twisted.test.test_reflect.Summer"),
Summer)
开发者ID:JohnDoes95,项目名称:project_parser,代码行数:7,代码来源:test_reflect.py
示例8: makeService
def makeService(self, options):
#
# Configure Memcached Client Pool
#
memcachepool.installPools(
config.Memcached.Pools,
config.Memcached.MaxClients,
)
multiService = service.MultiService()
notifiers = []
for key, settings in config.Notifications.Services.iteritems():
if settings["Enabled"]:
notifier = namedClass(settings["Service"])(settings)
notifier.setServiceParent(multiService)
notifiers.append(notifier)
internet.TCPServer(
config.Notifications.InternalNotificationPort,
InternalNotificationFactory(notifiers,
delaySeconds=config.Notifications.CoalesceSeconds),
interface=config.Notifications.BindAddress
).setServiceParent(multiService)
return multiService
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:27,代码来源:notify.py
示例9: main
def main():
# Setup the option parser
parser = optparse.OptionParser()
parser.add_option('-d', '--database', type="string", dest='database', help='database name')
parser.add_option('-u', '--user', type="string", dest='user', help='database user')
parser.add_option('-t', '--type', type="string", dest='type', help='item type')
parser.add_option('-e', '--execfile', type="string", dest='execfile', help='code to exec to get attrs')
# Parse the command line options
(options, args) = parser.parse_args()
if options.database is None or options.type is None:
parser.print_help()
return -1
# Extract the item type and turn the positional args into a dict of attrs
type = reflect.namedClass(options.type)
args = dict(arg.split('=', 1) for arg in args)
g = {}
execfile(options.execfile,g)
attrs=g['attrs']
# Create the object store
store = makeStore(database=options.database, user=options.user)
# Add the item and shutdown the reactor when it's complete
d = store.runInSession(createItem, type, args, attrs)
d.addCallbacks(itemCreated, error)
d.addBoth(lambda ignore: reactor.stop())
# Off we go
reactor.run()
sys.exit(exitCode)
开发者ID:timparkin,项目名称:into-the-light,代码行数:34,代码来源:createitem.py
示例10: test_namedClassLookup
def test_namedClassLookup(self):
"""
L{namedClass} should return the class object for the name it is passed.
"""
self.assertIdentical(
reflect.namedClass("twisted.python.reflect.Summer"),
reflect.Summer)
开发者ID:Alberto-Beralix,项目名称:Beralix,代码行数:7,代码来源:test_reflect.py
示例11: makeConfigurable
def makeConfigurable(self, cfgInfo, container, name):
"""Create a new configurable to a container, based on input from web form."""
cmd, args = string.split(cfgInfo, ' ', 1)
if cmd == "new": # create
obj = coil.createConfigurable(reflect.namedClass(args), container, name)
elif cmd == "dis": # dispense
methodHash = int(args)
if components.implements(container, coil.IConfigurator) and container.getType(name):
interface = container.getType(name)
elif components.implements(container, coil.IConfigCollection):
interface = container.entityType
else:
interface = None
for t in self.dispensers.getDispensers(interface):
obj, methodName, desc = t
if hash(t) == methodHash:
cfg = coil.getConfigurator(obj)
obj = getattr(cfg, methodName)()
print "created %s from dispenser" % obj
break
else:
raise ValueError, "Unrecognized command %r in cfgInfo %r" % (cmd, cfgInfo)
self.dispensers.addObject(obj)
return obj
开发者ID:fxia22,项目名称:ASM_xf,代码行数:25,代码来源:web.py
示例12: unjellyFromDOM_1
def unjellyFromDOM_1(self, unjellier, element):
from twisted.python.reflect import namedClass
self.integer = int(element.getAttribute("integer"))
self.instance = namedClass(element.getAttribute("instance"))()
self.name = element.getAttribute("name")
# just give us any ol' list
self.sequence = [self.instance, self.instance]
开发者ID:fxia22,项目名称:ASM_xf,代码行数:7,代码来源:test_persisted.py
示例13: opt_class
def opt_class(self, className):
"""A class that will be used to serve the root resource. Must implement hack.web2.iweb.IResource and take no arguments.
"""
if self['root']:
raise usage.UsageError("You may only have one root resource.")
classObj = reflect.namedClass(className)
self['root'] = iweb.IResource(classObj())
开发者ID:lzimm,项目名称:360io,代码行数:8,代码来源:tap.py
示例14: opt_processor
def opt_processor(self, proc):
"""`ext=class' where `class' is added as a Processor for files ending
with `ext'.
"""
if not isinstance(self['root'], static.File):
raise usage.UsageError("You can only use --processor after --path.")
ext, klass = proc.split('=', 1)
self['root'].processors[ext] = reflect.namedClass(klass)
开发者ID:Almad,项目名称:twisted,代码行数:8,代码来源:tap.py
示例15: tasks
def tasks(self):
"""
@return: list of class objects for each task defined in the current
configuration.
"""
allTasks = self.get('allTasks').split(',')
excludeTasks = self.get('excludeTasks').split(',')
return [ namedClass(t) for t in allTasks if not t in excludeTasks ]
开发者ID:Prash2504,项目名称:cqpython,代码行数:9,代码来源:task.py
示例16: main
def main(argv=None):
log.startLogging(file('child.log', 'w'))
if argv is None:
argv = sys.argv[1:]
if argv:
klass = reflect.namedClass(argv[0])
else:
klass = ConsoleManhole
runWithProtocol(klass)
开发者ID:0004c,项目名称:VTK,代码行数:10,代码来源:stdio.py
示例17: stepWithResult
def stepWithResult(self, result):
if self.doPostImport:
directory = directoryFromConfig(self.config)
# Load proxy assignments from XML if specified
if self.config.ProxyLoadFromFile:
proxydbClass = namedClass(self.config.ProxyDBService.type)
calendaruserproxy.ProxyDBService = proxydbClass(
**self.config.ProxyDBService.params)
loader = XMLCalendarUserProxyLoader(self.config.ProxyLoadFromFile)
yield loader.updateProxyDB()
# Populate the group membership cache
if (self.config.GroupCaching.Enabled and
self.config.GroupCaching.EnableUpdater):
proxydb = calendaruserproxy.ProxyDBService
if proxydb is None:
proxydbClass = namedClass(self.config.ProxyDBService.type)
proxydb = proxydbClass(**self.config.ProxyDBService.params)
updater = GroupMembershipCacheUpdater(proxydb,
directory,
self.config.GroupCaching.UpdateSeconds,
self.config.GroupCaching.ExpireSeconds,
self.config.GroupCaching.LockSeconds,
namespace=self.config.GroupCaching.MemcachedPool,
useExternalProxies=self.config.GroupCaching.UseExternalProxies)
yield updater.updateCache(fast=True)
uid, gid = getCalendarServerIDs(self.config)
dbPath = os.path.join(self.config.DataRoot, "proxies.sqlite")
if os.path.exists(dbPath):
os.chown(dbPath, uid, gid)
# Process old inbox items
self.store.setMigrating(True)
yield self.processInboxItems()
self.store.setMigrating(False)
# Migrate mail tokens from sqlite to store
yield migrateTokensToStore(self.config.DataRoot, self.store)
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:42,代码来源:upgrade.py
示例18: deadProperties
def deadProperties(self):
if not hasattr(self, "_dead_properties"):
# Get the property store from super
deadProperties = namedClass(config.RootResourcePropStoreClass)(self)
# Wrap the property store in a memory store
if isinstance(deadProperties, xattrPropertyStore):
deadProperties = CachingPropertyStore(deadProperties)
self._dead_properties = deadProperties
return self._dead_properties
开发者ID:eventable,项目名称:CalendarServer,代码行数:12,代码来源:root.py
示例19: makeService
def makeService(self, options):
"""Construct a server using MLLPFactory."""
from twisted.internet import reactor
from twistedhl7.mllp import IHL7Receiver, MLLPFactory
receiver_name = options['receiver']
receiver_class = reflect.namedClass(receiver_name)
verifyClass(IHL7Receiver, receiver_class)
factory = MLLPFactory(receiver_class())
endpoint = endpoints.serverFromString(reactor, options['endpoint'])
server = internet.StreamServerEndpointService(endpoint, factory)
server.setName(u"mllp-{0}".format(receiver_name))
return server
开发者ID:Brightmd,项目名称:twisted-hl7,代码行数:13,代码来源:mllp_plugin.py
示例20: directoryService
def directoryService(self):
"""
Get an appropriate directory service for this L{DBInspectService}'s
configuration, creating one first if necessary.
"""
if self._directory is None:
self._directory = directoryFromConfig(self.config)
proxydbClass = namedClass(config.ProxyDBService.type)
try:
calendaruserproxy.ProxyDBService = proxydbClass(**config.ProxyDBService.params)
except IOError:
print("Could not start proxydb service")
return self._directory
开发者ID:svn2github,项目名称:calendarserver-raw,代码行数:13,代码来源:dbinspect.py
注:本文中的twisted.python.reflect.namedClass函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论