本文整理汇总了Python中zope.component.getAdapter函数的典型用法代码示例。如果您正苦于以下问题:Python getAdapter函数的具体用法?Python getAdapter怎么用?Python getAdapter使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了getAdapter函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: _gotSession
def _gotSession(sess):
log.msg(
'bit.bot.http.socket: BotSocketProtocol.dataReceived._gotSession',
sess)
if sess:
getUtility(ISockets).add('bot', sessionid, token, self)
request = queryAdapter(self, ISocketRequest, name=data['request'])
if data['request'] == 'message':
message = data['message']
if message.startswith('>'):
request = getAdapter(self, ISocketRequest, name="command")
message = message[1:]
data['command'] = message.strip().split(' ')[0]
data['args'] = ' '.join(message.strip().split(' ')[1:])
elif message.startswith('~'):
request = getAdapter(
self, ISocketRequest, name="subscribe")
data['subscribe'] = message[1:]
else:
request = getAdapter(
self, ISocketRequest, name="message")
else:
request = queryAdapter(
self, ISocketRequest, name=data['request'])
if request:
request.load(sessionid, sess, data)
else:
print 'NO REQUEST ADAPTER FOR: %s' % data['request']
开发者ID:bithub,项目名称:bit.bot.socket,代码行数:32,代码来源:socket.py
示例2: create_queue
def create_queue(self, queue):
# Check to see if the queue exists.
try:
self._checking = True
yield getAdapter(self.chan, IAMQPChannelAdapter).declareQueue(queue, True)
return
except ChannelClosedError as e:
log.debug(("Channel {0} doesn't exist, attempting to create it.").format(queue.name))
self.chan = yield self.get_channel()
finally:
self._checking = False
# Declare the queue
try:
yield getAdapter(self.chan, IAMQPChannelAdapter).declareQueue(queue, False)
except ChannelClosedError as e:
# Here we handle the case where we redeclare a queue
# with different properties. When this happens, Rabbit
# both returns an error and closes the channel. We
# need to detect this and reopen the channel, since
# the existing queue will work fine (although it will
# not use the modified config).
if e.replyCode == 406:
# PRECONDITION_FAILED -- properties changed
# Remove the channel and allow it to be reopened
log.warn(("Attempted to redeclare queue {0} with "
"different arguments. You will need to "
"delete the queue to pick up the new "
"configuration.").format(queue.name))
log.debug(e)
self.chan = yield self.get_channel()
else:
raise
开发者ID:zenoss,项目名称:zenoss-protocols,代码行数:33,代码来源:amqp.py
示例3: renewAlephRecords
def renewAlephRecords(wfStateInfo):
logger.info("renewAlephRecords")
print "renew Aleph Records"
context = wfStateInfo.object
with api.env.adopt_user(username="system"):
getAdapter(context, IAMQPSender, name="load-aleph-records-by-title").send()
getAdapter(context, IAMQPSender, name="renew-aleph-records-by-icz-sysnumber").send()
开发者ID:edeposit,项目名称:edeposit.policy,代码行数:7,代码来源:eperiodical_wf_scripts.py
示例4: get_node_view
def get_node_view(node, action='', name='', ext=''):
if ext:
view_interface = getUtility(IViewExtFactory, ext).interface
else:
view_interface = INodeView
if isinstance(node, ViewResults):
return getAdapter(node, view_interface)
view = None
if name:
try:
factory = getUtility(INodeFactory, name)
except ComponentLookupError:
# /node/action/name/ - dynamic action, static name
view = getMultiAdapter((node, action), view_interface, name)
else:
# /node/action/factory/ - static action
view = getMultiAdapter((node, factory), view_interface, action)
else:
try:
# /node/action/ - static action
view = getAdapter(node, view_interface, action)
except ComponentLookupError:
# /node/action/ - dynamic action
view = getMultiAdapter((node, action), view_interface)
return view
开发者ID:ManoSeimas,项目名称:django-sboard,代码行数:28,代码来源:views.py
示例5: submitSysNumbersSearch
def submitSysNumbersSearch(wfStateInfo):
logger.info("submitSysNumberSearch")
print "submit sysnumber search"
originalfile = wfStateInfo.object
# epublication = aq_parent(aq_inner(originalfile))
with api.env.adopt_user(username="system"):
getAdapter(originalfile, IAMQPSender, name="sysnumber-aleph-search").send()
开发者ID:edeposit,项目名称:edeposit.policy,代码行数:7,代码来源:originalfile_wf_scripts.py
示例6: _addSortUi
def _addSortUi(self, header, column):
columnName = column.name
resource_path = component.getAdapter(self.request, name='zc.table')()
if (interfaces.IColumnSortedItems.providedBy(self.items) and
self.items.sort_on):
sortColumnName, sortReversed = self.items.sort_on[0]
else:
sortColumnName = sortReversed = None
if columnName == sortColumnName:
path = component.getAdapter(self.request, name='zc.table')()
if sortReversed:
dirIndicator = ('<img src="%s/sort_arrows_up.gif" '
'style="vertical-align: top; '
'margin-top: 0px" '
'alt="(ascending)"/>' % resource_path)
else:
dirIndicator = ('<img src="%s/sort_arrows_down.gif" '
'style="vertical-align: top; '
'margin-top: 0px" '
'alt="(descending)"/>' % resource_path)
else:
dirIndicator = ('<img src="%s/sort_arrows.gif" '
'style="vertical-align: top; '
'margin-top: 0px" '
'alt="(sortable)"/>' % resource_path)
sort_on_name = getSortOnName(self.prefix)
script_name = self.script_name
return self._header_template(locals())
开发者ID:BGCX261,项目名称:zope-alchemist-svn-to-git,代码行数:28,代码来源:table.py
示例7: submitThumbnailGenerating
def submitThumbnailGenerating(wfStateInfo):
logger.info("submitThumbnailGenerating")
print "submit thumbnail generating"
originalfile = wfStateInfo.object
epublication = aq_parent(aq_inner(originalfile))
with api.env.adopt_user(username="system"):
getAdapter(originalfile, IAMQPSender, name="generate-thumbnail").send()
开发者ID:edeposit,项目名称:edeposit.policy,代码行数:7,代码来源:originalfile_wf_scripts.py
示例8: test_many_groups
def test_many_groups(self):
getAdapter(self.portal, IUserGroupsSettingsSchema).set_many_groups(True)
self.assertEqual(
getAdapter(
self.portal, IUserGroupsSettingsSchema).get_many_groups(),
True
)
开发者ID:JensenJiang,项目名称:Products.CMFPlone,代码行数:7,代码来源:test_controlpanel_bbb_usergroups_adapter.py
示例9: _bind
def _bind(self):
queueConfig = self._queueSchema.getQueue(self._queueName)
try:
getAdapter(self.channel, IAMQPChannelAdapter).declareQueue(queueConfig)
except ChannelClosedError as e:
# Here we handle the case where we redeclare a queue
# with different properties. When this happens, Rabbit
# both returns an error and closes the channel. We
# need to detect this and reopen the channel, since
# the existing queue will work fine (although it will
# not use the modified config).
if e.replyCode == 406:
# PRECONDITION_FAILED -- properties changed
# Remove the channel and allow it to be reopened
log.warn(("Attempted to redeclare queue {0} with "
"different arguments. You will need to "
"delete the queue to pick up the new "
"configuration.").format(queueConfig.name))
log.debug(e)
self._channel = None
else:
raise
for outboundExchange in self._exchanges:
exchangeConfig = self._queueSchema.getExchange(outboundExchange)
getAdapter(self.channel, IAMQPChannelAdapter).declareExchange(exchangeConfig)
开发者ID:malysoun,项目名称:zenoss-protocols,代码行数:27,代码来源:amqp.py
示例10: test_properties_adapter_setter
def test_properties_adapter_setter(self):
adapter1 = getAdapter(self.layer['testdoc1'], IDataCollector,
name="properties_data_adapter")
data = adapter1.getData()
# setter
adapter2 = getAdapter(self.layer['testdoc2'], IDataCollector,
name="properties_data_adapter")
adapter2.setData(data, metadata=None)
# test boolean
self.assertEquals(getattr(self.layer['testdoc2'], 'bool'), True)
# test date
self.assertEquals(getattr(self.layer['testdoc2'], 'date'),
DateTime('2000/01/01'))
# test float
self.assertEquals(getattr(self.layer['testdoc2'], 'float'),
2.1000000000000001)
# test int
self.assertEquals(getattr(self.layer['testdoc2'], 'int'), 2)
# test lines
self.assertEquals(getattr(self.layer['testdoc2'], 'lines')[0], 'row1')
self.assertEquals(getattr(self.layer['testdoc2'], 'lines')[1], 'row2')
# test string
self.assertEquals(getattr(self.layer['testdoc2'], 'string'),
'Hello World!')
开发者ID:4teamwork,项目名称:ftw.publisher.core,代码行数:25,代码来源:test_properties_adapter.py
示例11: url
def url(view):
helper = getAdapter(context, IExposureSourceAdapter)
exposure, workspace, path = helper.source()
note = getAdapter(context, name=view)
filename = note.filename or path
return 'opencor://openFile/%s/rawfile/%s/%s' % (
workspace.absolute_url(), exposure.commit_id, filename)
开发者ID:PMR2,项目名称:cellml.pmr2,代码行数:7,代码来源:util.py
示例12: __init__
def __init__(self, context, request):
super(ModerationView, self).__init__(context, request)
self.mem_list = IWriteMembershipList(context)
annot = IAnnotations(self.context)
self.listen_annot = annot.setdefault(PROJECTNAME, OOBTree())
self.mod_post_pending_list = getAdapter(context, IPostPendingList, 'pending_mod_post')
self.pmod_post_pending_list = getAdapter(context, IPostPendingList, 'pending_pmod_post')
self.sub_pending_list = getAdapter(context, IMembershipPendingList, 'pending_sub_mod_email')
开发者ID:socialplanning,项目名称:opencore-listen,代码行数:8,代码来源:moderation.py
示例13: __call__
def __call__(self):
sub_action = self.request.get('subscribe_member', None)
unsub_action = self.request.get('unsubscribe_member', None)
email_action = self.request.get('subscribe_email', None)
self.request.set('enable_border', True)
self.errors = errors = {}
logged_in_mem = self._get_logged_in_user()
self.user_logged_in = False
if logged_in_mem:
self.user_email = lookup_email(logged_in_mem.getId(), self.context)
self.user_logged_in = True
else:
#XXX what should this be?
self.user_email = ''
self.mem_list = IWriteMembershipList(self.context)
# the appropriate sub_policy needs to be instantiated
# depending on list type
self.sub_policy = getAdapter(self.context, IUserTTWMembershipPolicy)
if sub_action:
self.subscribe()
elif unsub_action:
self.unsubscribe()
elif email_action:
address = self.request.get('email_address', None)
if not address:
errors['email_address'] = _('An email address is required')
elif EMAIL_RE.match(address) is None:
errors['email_address'] = _('This email address is invalid')
elif self.mem_list.is_subscribed(address):
errors['email_address'] = \
_('This email address is already subscribed')
else:
# everything is OK, send a request mail the
# appropriate sub_policy needs to be instantiated
# depending on list type
sub_policy_for_email = getAdapter(self.context, IUserEmailMembershipPolicy)
ret = sub_policy_for_email.enforce({'email':address,
'subject':'subscribe'})
if ret == MEMBERSHIP_ALLOWED:
# make user a subscriber
self.mem_list.subscribe(address)
self.request.set('portal_status_message', 'Email subscribed')
elif ret == MEMBERSHIP_DEFERRED:
self.request.set('portal_status_message',
'Subscription request sent')
else:
self.request.set('portal_status_message', 'Bad email address')
# Blank the email field to avoid the postback
self.request.set('email_address', '')
self.request.set('subscribe_email', '')
return self.index()
开发者ID:socialplanning,项目名称:opencore-listen,代码行数:58,代码来源:mailinglist_views.py
示例14: test_reference_number_works_for_objects_marked_as_BaseDocument
def test_reference_number_works_for_objects_marked_as_BaseDocument(self):
d1 = createContentInContainer(self.portal, 'opengever.document.document')
b1 = createContentInContainer(self.portal, 'SimpleDocument')
d2 = createContentInContainer(self.portal, 'opengever.document.document')
self.assertEquals(['Client1 / 1', 'Client1 / 2', 'Client1 / 3'],
[getAdapter(d1, IReferenceNumber).get_number(),
getAdapter(b1, IReferenceNumber).get_number(),
getAdapter(d2, IReferenceNumber).get_number()])
开发者ID:4teamwork,项目名称:opengever.core,代码行数:9,代码来源:test_document.py
示例15: test_get_default_language
def test_get_default_language(self):
self.assertEqual(
getAdapter(self.portal, ILanguageSchema).default_language,
'en'
)
self.settings.default_language = 'de'
self.assertEquals(
getAdapter(self.portal, ILanguageSchema).default_language,
'de'
)
开发者ID:FemtoSistemas,项目名称:Products.CMFPlone,代码行数:10,代码来源:test_controlpanel_bbb_language_adapter.py
示例16: test_get_available_languages
def test_get_available_languages(self):
self.assertEqual(
getAdapter(self.portal, ILanguageSchema).available_languages,
['en']
)
self.settings.available_languages = ['en', 'de']
self.assertEquals(
getAdapter(self.portal, ILanguageSchema).available_languages,
['en', 'de']
)
开发者ID:FemtoSistemas,项目名称:Products.CMFPlone,代码行数:10,代码来源:test_controlpanel_bbb_language_adapter.py
示例17: test_get_show_excluded_items
def test_get_show_excluded_items(self):
self.assertEqual(
getAdapter(self.portal, INavigationSchema).show_excluded_items,
True
)
self.navigation_settings.show_excluded_items = False
self.assertEqual(
getAdapter(self.portal, INavigationSchema).show_excluded_items,
False
)
开发者ID:erral,项目名称:Products.CMFPlone,代码行数:10,代码来源:test_controlpanel_bbb_navigation_adapter.py
示例18: test_get_nonfolderish_tabs
def test_get_nonfolderish_tabs(self):
self.assertEqual(
getAdapter(self.portal, INavigationSchema).nonfolderish_tabs,
True
)
self.navigation_settings.nonfolderish_tabs = False
self.assertEquals(
getAdapter(self.portal, INavigationSchema).nonfolderish_tabs,
False
)
开发者ID:erral,项目名称:Products.CMFPlone,代码行数:10,代码来源:test_controlpanel_bbb_navigation_adapter.py
示例19: book_return
def book_return(self, book_barcode):
book = Book()
book.number = book_barcode
bookdboperation = getAdapter(book, IDbOperation)
book = bookdboperation.get()[0]
circulation = Circulation()
circulation.book = book
circulationdboperation = getAdapter(circulation, IDbOperation)
circulationdboperation.delete()
开发者ID:eugeneai,项目名称:ZCA,代码行数:10,代码来源:vc.py
示例20: test_get_lock_on_ttw_edit_setting
def test_get_lock_on_ttw_edit_setting(self):
self.assertEqual(
getAdapter(self.portal, IEditingSchema).lock_on_ttw_edit,
True
)
self.settings.lock_on_ttw_edit = False
self.assertEqual(
getAdapter(self.portal, IEditingSchema).lock_on_ttw_edit,
False
)
开发者ID:JensenJiang,项目名称:Products.CMFPlone,代码行数:10,代码来源:test_controlpanel_bbb_editing_adapter.py
注:本文中的zope.component.getAdapter函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论