我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用zope.interface.providedBy()。
def test_no_adapts(self): from zope.interface import Interface from zope.interface import implementer from zope.interface import providedBy from guillotina.component.globalregistry import get_global_components from guillotina.component._declaration import adapter class IFoo(Interface): pass @implementer(IFoo) class Foo(object): pass @adapter(IFoo) def _handler(context): assert 0, "DON'T GO HERE" self._callFUT(_handler) gsm = get_global_components() regs = list(gsm.registeredHandlers()) self.assertEqual(len(regs), 1) hr = regs[0] self.assertEqual(list(hr.required), list(providedBy(Foo()))) self.assertEqual(hr.name, '') self.assertTrue(hr.factory is _handler)
def test_w_factory_returning_spec(self): from zope.interface import Interface from zope.interface import implementer from zope.interface import providedBy from guillotina.component.interfaces import IFactory class IFoo(Interface): pass class IBar(Interface): pass @implementer(IFoo, IBar) class _Factory(object): def get_interfaces(self): return providedBy(self) _factory = _Factory() class Context(object): def __conform__(self, iface): return self def getUtilitiesFor(self, iface): if iface is IFactory: return [('test', _factory)] self.assertEqual(list(self._callFUT(IFoo, context=Context())), [('test', _factory)]) self.assertEqual(list(self._callFUT(IBar, context=Context())), [('test', _factory)])
def query_multi_adapter(objects, interface=Interface, name=_BLANK, default=None, context=None, args=[], kwargs={}): try: registry = get_component_registry(context) except ComponentLookupError: # Oh blast, no site manager. This should *never* happen! return default factory = registry.adapters.lookup(map(providedBy, objects), interface, name) if factory is None: return default result = factory(*objects, *args, **kwargs) if result is None: return default return result
def __get__(self, inst, owner): name, check = self._info default_factory = inst.__dict__.get('defaultFactory') # If there is no default factory, simply return the default. if default_factory is None: return inst.__dict__[name] # Get the default value by calling the factory. Some factories might # require a context to produce a value. if IContextAwareDefaultFactory.providedBy(default_factory): value = default_factory(inst.context) else: value = default_factory() # Check that the created value is valid. if check is not None: check(inst, value) elif value != inst.missing_value: inst.validate(value) return value
def __eq__(self, other): # should be the same type if type(self) != type(other): return False # should have the same properties names = {} # used as set of property names, ignoring values for interface in providedBy(self): names.update(get_fields(interface)) # order will be different always, don't compare it if 'order' in names: del names['order'] for name in names: if getattr(self, name) != getattr(other, name): return False return True
def from_unicode(self, str): """ >>> b = Bool() >>> IFromUnicode.providedBy(b) True >>> b.from_unicode('True') True >>> b.from_unicode('') False >>> b.from_unicode('true') True >>> b.from_unicode('false') or b.from_unicode('False') False """ v = str == 'True' or str == 'true' self.validate(v) return v
def test_classImplements_base_not_derived(self): from zope.interface import Interface from zope.interface import implementedBy from zope.interface import providedBy class IBase(Interface): def method(): pass class IDerived(IBase): pass class Current(): __implemented__ = IBase def method(self): pass current = Current() self.assertTrue(IBase.implementedBy(Current)) self.assertFalse(IDerived.implementedBy(Current)) self.assertTrue(IBase in implementedBy(Current)) self.assertFalse(IDerived in implementedBy(Current)) self.assertTrue(IBase in providedBy(current)) self.assertFalse(IDerived in providedBy(current))
def render_panel(context, request, name='', *args, **kw): """Render a panel. TODO: Clean this up Renders the named panel, returning a `unicode` object that is the rendered HTML for the panel. The panel is looked up using the current context (or the context given as keyword argument, to override the context in which the panel is called) and an optional given name (which defaults to an empty string). The panel is called passing in the current context, request and any additional parameters passed into the `render_panel` call. In case a panel isn't found, `None` is returned. """ if not "controls" in kw: kw["controls"] = True adapters = request.registry.adapters panel = adapters.lookup((providedBy(context),), IPanel, name=name) assert panel, "Could not find panel {}".format(name) return Markup(panel(context, request, *args, **kw))
def requestAvatarId(self, credentials): """ Part of the L{ICredentialsChecker} interface. Called by a portal with some credentials to check if they'll authenticate a user. We check the interfaces that the credentials provide against our list of acceptable checkers. If one of them matches, we ask that checker to verify the credentials. If they're valid, we call our L{_cbGoodAuthentication} method to continue. @param credentials: the credentials the L{Portal} wants us to verify """ ifac = providedBy(credentials) for i in ifac: c = self.checkers.get(i) if c is not None: d = defer.maybeDeferred(c.requestAvatarId, credentials) return d.addCallback(self._cbGoodAuthentication, credentials) return defer.fail(UnhandledCredentials("No checker for %s" % \ ', '.join(map(reflect.qual, ifac))))
def makeConnection(self, transport): """ Connect this wrapper to the given transport and initialize the necessary L{OpenSSL.SSL.Connection} with a memory BIO. """ self._tlsConnection = self.factory._createConnection(self) self._appSendBuffer = [] # Add interfaces provided by the transport we are wrapping: for interface in providedBy(transport): directlyProvides(self, interface) # Intentionally skip ProtocolWrapper.makeConnection - it might call # wrappedProtocol.makeConnection, which we want to make conditional. Protocol.makeConnection(self, transport) self.factory.registerProtocol(self) if self._connectWrapped: # Now that the TLS layer is initialized, notify the application of # the connection. ProtocolWrapper.makeConnection(self, transport) # Now that we ourselves have a transport (initialized by the # ProtocolWrapper.makeConnection call above), kick off the TLS # handshake. self._checkHandshakeStatus()
def _checkHandshakeStatus(self): """ Ask OpenSSL to proceed with a handshake in progress. Initially, this just sends the ClientHello; after some bytes have been stuffed in to the C{Connection} object by C{dataReceived}, it will then respond to any C{Certificate} or C{KeyExchange} messages. """ # The connection might already be aborted (eg. by a callback during # connection setup), so don't even bother trying to handshake in that # case. if self._aborted: return try: self._tlsConnection.do_handshake() except WantReadError: self._flushSendBIO() except Error: self._tlsShutdownFinished(Failure()) else: self._handshakeDone = True if IHandshakeListener.providedBy(self.wrappedProtocol): self.wrappedProtocol.handshakeCompleted()
def _applyProtocolNegotiation(self, connection): """ Applies ALPN/NPN protocol neogitation to the connection, if the factory supports it. @param connection: The OpenSSL connection object to have ALPN/NPN added to it. @type connection: L{OpenSSL.SSL.Connection} @return: Nothing @rtype: L{None} """ if IProtocolNegotiationFactory.providedBy(self.wrappedFactory): protocols = self.wrappedFactory.acceptableProtocols() context = connection.get_context() _setAcceptableProtocols(context, protocols) return
def requestAvatarId(self, credentials): ifac = interface.providedBy(credentials) for i in ifac: c = self.checkers.get(i) if c is not None: return c.requestAvatarId(credentials).addCallback( self._cbGoodAuthentication, credentials) return defer.fail(UnhandledCredentials("No checker for %s" % \ ', '.join(map(reflect.qal, ifac))))
def _generateCacheEntry(provider): dropin = CachedDropin(provider.__name__, provider.__doc__) for k, v in provider.__dict__.iteritems(): plugin = IPlugin(v, None) if plugin is not None: cachedPlugin = CachedPlugin(dropin, k, v.__doc__, list(providedBy(plugin))) return dropin
def login(self, credentials, mind, *interfaces): """ @param credentials: an implementor of twisted.cred.credentials.ICredentials @param mind: an object which implements a client-side interface for your particular realm. In many cases, this may be None, so if the word 'mind' confuses you, just ignore it. @param interfaces: list of interfaces for the perspective that the mind wishes to attach to. Usually, this will be only one interface, for example IMailAccount. For highly dynamic protocols, however, this may be a list like (IMailAccount, IUserChooser, IServiceInfo). To expand: if we are speaking to the system over IMAP, any information that will be relayed to the user MUST be returned as an IMailAccount implementor; IMAP clients would not be able to understand anything else. Any information about unusual status would have to be relayed as a single mail message in an otherwise-empty mailbox. However, in a web-based mail system, or a PB-based client, the ``mind'' object inside the web server (implemented with a dynamic page-viewing mechanism such as woven) or on the user's client program may be intelligent enough to respond to several ``server''-side interfaces. @return: A deferred which will fire a tuple of (interface, avatarAspect, logout). The interface will be one of the interfaces passed in the 'interfaces' argument. The 'avatarAspect' will implement that interface. The 'logout' object is a callable which will detach the mind from the avatar. It must be called when the user has conceptually disconnected from the service. Although in some cases this will not be in connectionLost (such as in a web-based session), it will always be at the end of a user's interactive session. """ ifac = providedBy(credentials) for i in ifac: c = self.checkers.get(i) if c is not None: return maybeDeferred(c.requestAvatarId, credentials ).addCallback(self.realm.requestAvatar, mind, *interfaces ) return defer.fail(failure.Failure(error.UnhandledCredentials( "No checker for %s" % ', '.join(map(reflect.qual, ifac)))))
def makeConnection(self, transport): directlyProvides(self, *providedBy(self) + providedBy(transport)) Protocol.makeConnection(self, transport) # Transport relaying
def testUtil(self): self.assert_(IC in implementedBy(C)) self.assert_(I1 in implementedBy(A)) self.assert_(not I1 in implementedBy(C)) self.assert_(I2 in implementedBy(B)) self.assert_(not I2 in implementedBy(C)) self.assert_(IC in providedBy(C())) self.assert_(I1 in providedBy(A())) self.assert_(not I1 in providedBy(C())) self.assert_(I2 in providedBy(B())) self.assert_(not I2 in providedBy(C()))
def testVerifyImplementation(self): from zope.interface.verify import verifyClass self.assert_(verifyClass(FooInterface, Foo)) self.assert_(Interface.providedBy(I1))
def testIssue228(self): # Test for http://collector.zope.org/Zope3-dev/228 class I(Interface): "xxx" class Bad: __providedBy__ = None # Old style classes don't have a '__class__' attribute self.failUnlessRaises(AttributeError, I.providedBy, Bad)
def test_directlyProvides(self): class IA1(Interface): pass class IA2(Interface): pass class IB(Interface): pass class IC(Interface): pass class A(Odd): implements(IA1, IA2) class B(Odd): implements(IB) class C(A, B): implements(IC) ob = C() directlyProvides(ob, I1, I2) self.assert_(I1 in providedBy(ob)) self.assert_(I2 in providedBy(ob)) self.assert_(IA1 in providedBy(ob)) self.assert_(IA2 in providedBy(ob)) self.assert_(IB in providedBy(ob)) self.assert_(IC in providedBy(ob)) directlyProvides(ob, directlyProvidedBy(ob)-I2) self.assert_(I1 in providedBy(ob)) self.failIf(I2 in providedBy(ob)) self.failIf(I2 in providedBy(ob)) directlyProvides(ob, directlyProvidedBy(ob), I2) self.assert_(I2 in providedBy(ob))
def adapter_hook(self, provided, object, name=u'', default=None): required = providedBy(object) cache = self._getcache(provided, name) factory = cache.get(required, _not_in_mapping) if factory is _not_in_mapping: factory = self.lookup((required, ), provided, name) if factory is not None: result = factory(object) if result is not None: return result return default
def subscribers(self, objects, provided): subscriptions = self.subscriptions(map(providedBy, objects), provided) if provided is None: result = () for subscription in subscriptions: subscription(*objects) else: result = [] for subscription in subscriptions: subscriber = subscription(*objects) if subscriber is not None: result.append(subscriber) return result
def get_obj_value(obj, fields={}): interfaces = [i for i in providedBy(obj).interfaces()] if not interfaces: return {} result = {} objects_to_add = [] if fields: for interface in interfaces: attributes = [a for a in interface if a in fields and isinstance(interface[a], Attribute)] for attr in attributes: result[attr], to_add = get_attr_value( interface[attr], obj, fields.get(attr, {})) objects_to_add.extend(to_add) else: for interface in interfaces: attributes = [a for a in interface if isinstance(interface[a], Attribute)] for attr in attributes: result[attr], to_add = get_attr_value(interface[attr], obj) objects_to_add.extend(to_add) result = normalize_value(result) result['@id'] = '_:' + str(get_oid(obj, 'None')) result['@type'] = getattr( obj, 'type_title', obj.__class__.__name__) return result, objects_to_add
def asubscribers(self, objects, provided): subscriptions = self.subscriptions(map(providedBy, objects), provided) if provided is None: result = () for subscription in subscriptions: if asyncio.iscoroutinefunction(subscription): await subscription(*objects) else: result = [] for subscription in subscriptions: if asyncio.iscoroutinefunction(subscription): subscriber = await subscription(*objects) if subscriber is not None: result.append(subscriber) return result
def subscribers(self, objects, provided): subscriptions = self.subscriptions(map(providedBy, objects), provided) if provided is None: result = () for subscription in subscriptions: if not asyncio.iscoroutinefunction(subscription): subscription(*objects) else: result = [] for subscription in subscriptions: if not asyncio.iscoroutinefunction(subscription): subscriber = subscription(*objects) if subscriber is not None: result.append(subscriber) return result
def test_gsm_is_IComponentLookup(self): from guillotina.component.globalregistry import base from guillotina.component.interfaces import IComponentLookup gsm = self._callFUT() self.assertTrue(gsm is base) self.assertTrue(IComponentLookup.providedBy(gsm))
def test_sm_is_IComponentLookup(self): from guillotina.component.interfaces import IComponentLookup sm = self._callFUT() self.assertTrue(IComponentLookup.providedBy(sm))
def asubscribers(self, objects, provided): subscriptions = self.subscriptions(map(providedBy, objects), provided) results = [] for subscription in subscriptions: if asyncio.iscoroutinefunction(subscription): results.append(await subscription(*objects)) return results
def subscribers(self, objects, provided): subscriptions = self.subscriptions(map(providedBy, objects), provided) result = [] for subscription in subscriptions: if not asyncio.iscoroutinefunction(subscription): result.append(subscription(*objects)) return result
def utility(_context, provides=None, component=None, factory=None, name=''): if factory and component: raise TypeError("Can't specify factory and component.") if provides is None: if factory: provides = list(implementedBy(factory)) else: provides = list(providedBy(component)) if len(provides) == 1: provides = provides[0] else: raise TypeError("Missing 'provides' attribute") if name == '': if factory: name = getName(factory) else: name = getName(component) _context.action( discriminator=('utility', provides, name), callable=handler, args=('registerUtility', component, provides, name), kw=dict(factory=factory)) _context.action( discriminator=None, callable=provide_interface, args=('', provides))
def __provides__(self): return providedBy(self.field) + implementedBy(FieldReadAccessor)
def test___provides___w_field_no_provides(self): from zope.interface import implementedBy from zope.interface import providedBy wrapped = self._makeOne(object()) self.assertEqual(list(providedBy(wrapped)), list(implementedBy(self._getTargetClass())))
def makeConnection(self, transport): """ Upon connection, provides the transport interface, and forwards ourself as the transport to C{self.wrappedProtocol}. @type transport: L{twisted.internet.interfaces.ITransport} provider. @param transport: The transport to use for the protocol. """ directlyProvides(self, providedBy(transport)) WebSocketsProtocol.makeConnection(self, transport) self.wrappedProtocol.makeConnection(self)
def test_providedBy_miss(self): from zope.interface import interface from zope.interface.declarations import _empty sb = self._makeOne() def _providedBy(obj): return _empty with _Monkey(interface, providedBy=_providedBy): self.assertFalse(sb.providedBy(object()))
def test_providedBy_hit(self): from zope.interface import interface sb = self._makeOne() class _Decl(object): _implied = {sb: {},} def _providedBy(obj): return _Decl() with _Monkey(interface, providedBy=_providedBy): self.assertTrue(sb.providedBy(object()))
def _makeOne(self, object_should_provide): class IB(self._getTargetClass()): def _call_conform(self, conform): return conform(self) def providedBy(self, obj): return object_should_provide return IB()
def test_classImplements_simple(self): from zope.interface import Interface from zope.interface import implementedBy from zope.interface import providedBy class ICurrent(Interface): def method1(a, b): pass def method2(a, b): pass class IOther(Interface): pass class Current(object): __implemented__ = ICurrent def method1(self, a, b): return 1 def method2(self, a, b): return 2 current = Current() self.assertTrue(ICurrent.implementedBy(Current)) self.assertFalse(IOther.implementedBy(Current)) self.assertTrue(ICurrent in implementedBy(Current)) self.assertFalse(IOther in implementedBy(Current)) self.assertTrue(ICurrent in providedBy(current)) self.assertFalse(IOther in providedBy(current))
def test_classImplements_multiple(self): from zope.interface import Interface from zope.interface import implementedBy from zope.interface import providedBy class ILeft(Interface): def method(): pass class IRight(ILeft): pass class Left(object): __implemented__ = ILeft def method(self): pass class Right(object): __implemented__ = IRight class Ambi(Left, Right): pass ambi = Ambi() self.assertTrue(ILeft.implementedBy(Ambi)) self.assertTrue(IRight.implementedBy(Ambi)) self.assertTrue(ILeft in implementedBy(Ambi)) self.assertTrue(IRight in implementedBy(Ambi)) self.assertTrue(ILeft in providedBy(ambi)) self.assertTrue(IRight in providedBy(ambi))
def test_interface_object_provides_Interface(self): from zope.interface import Interface class AnInterface(Interface): pass self.assertTrue(Interface.providedBy(AnInterface))