How to use Zope - 10 common examples

class ReceiverProtocol(Protocol):
            def __init__(self, finished):
                self.finished = finished
                self.body = []

            def dataReceived(self, bytes):
                self.body.append(bytes)

            def connectionLost(self, reason):
                self.finished.callback(''.join(self.body))


        class StringProducer(object):
            implements(IBodyProducer)

            def __init__(self, body):
                self.body = body
                self.length = len(body)

            def startProducing(self, consumer):
                consumer.write(self.body)
                return defer.succeed(None)

            def pauseProducing(self):
                pass

            def stopProducing(self):
                pass

from plone.app.portlets.portlets import base
from plone.portlets.interfaces import IPortletDataProvider
from zope.interface import implementer

import logging
import warnings


logging.captureWarnings(True)


class ILatestSectionableNITFPortlet(IPortletDataProvider):
    """BBB."""


@implementer(ILatestSectionableNITFPortlet)
class Assignment(base.Assignment):
    """BBB."""


class Renderer(base.Renderer):
    """BBB."""

    def render(self):
        msg = (
            'Use of portlet "Latest Sectionable NITF" is deprecated; '
            'remove assignment manually from {0}'.format(self.context))
        warnings.warn(msg, DeprecationWarning)


class AddForm(base.AddForm):
    """BBB."""

self.result.log.append((ts, self.peerIndex, sessionId, msg.encode('utf8')))
      return ts

   def clientConnectionLost(self, connector, reason):
      reason = str(reason.value)
      self.log("Client connection lost: %s" % reason)
      self.onGone.callback(None)

   def clientConnectionFailed(self, connector, reason):
      reason = str(reason.value)
      self.log("Client connection failed: %s" % reason)
      self.onGone.callback(reason)



@implementer(ITestCase)
class WampCase:

   factory = None
   index = None
   description = None
   expectation = None
   params = None


   def __init__(self, testee, spec):
      self.testee = testee
      self.spec = spec

      self._uriSuffix = '#' + str(random.randint(0, 1000000))

      if self.testee.options.has_key('rtt'):

True

Unfortunately ftests don't have good isolation from each other yet.
"""

from zope.interface import Interface
from zope import schema

import grok
from grok import index

class Herd(grok.Container, grok.Application):
    pass

class IMammoth(Interface):
    name = schema.TextLine(title=u'Name')

class IMammoth2(Interface):
    name = schema.TextLine(title=u'Name')

class MammothIndexes(grok.Indexes):
    grok.site(Herd)
    grok.context(IMammoth)

    name = index.Field()

class MammothIndexes2(grok.Indexes):
    grok.site(Herd)
    grok.context(IMammoth2)

    name = index.Field()

@implementer(IKeyState)
class SphinxNodeKeyState:

    def __init__(self, public_key, private_key):
        self.public_key = public_key
        self.private_key = private_key

    def get_private_key(self):
        return self.private_key

    def get_public_key(self):
        return self.public_key

@zope.interface.implementer(IReader)
class FixedNoiseReader():

    def __init__(self, hexed_noise):
        self.noise = binascii.unhexlify(hexed_noise)
        self.count = 0
        self.fallback = RandReader()

    def read(self, n):
        if n > len(self.noise):
            print("%s > %s" % (n, len(self.noise)))
            return self.fallback.read(n)
        ret = self.noise[:n]
        self.noise = self.noise[n:]
        self.count += n
        return ret

def contents_content(tc):
        tc.loginAsPortalOwner()
        proj = tc.portal.projects.p2
        proj.invokeFactory('Document', 'new1', title='new title')
        proj.invokeFactory('Image', 'img1', title='new image')
        proj.restrictedTraverse('project-home').invokeFactory('FileAttachment', 'fa1', title='new file')
        proj.new1.invokeFactory('FileAttachment', 'fa2', title='new1 file')
        proj.invokeFactory('Folder', 'lists', title='Listen Stub')
        lists = proj.lists
        lists.setLayout('mailing_lists')
        alsoProvides(lists, IListenContainer)
        enableLocalSiteHook(tc.portal)
        setSite(tc.portal)
        setHooks()
        proj.lists.invokeFactory('Open Mailing List', 'list1', title=u'new list')

def _getPropertiesForVariants(self, cart_item):
        """
        """        
        u = getUtility(INumberConverter)
        cm = ICurrencyManagement(self.context)

        variant = cart_item.getProduct()
        product = variant.aq_inner.aq_parent

        selected_options = {}
        for property in variant.getForProperties():
            name, value = property.split(":")
            selected_options[name] = value
            
        pm = IPropertyManagement(product)
        
        result = []
        for property in pm.getProperties():
            
            # Only properties with at least one option are displayed.

from zope.interface import implementer
from zope.schema import getFieldsInOrder

import random


class ICollectionTile(IPersistentCoverTile):

    header = schema.TextLine(
        title=_(u'Header'),
        required=False,
    )

    form.omitted('title')
    form.no_omit(IDefaultConfigureForm, 'title')
    title = schema.TextLine(
        title=_(u'Title'),
        required=False,
    )

    form.omitted('description')
    form.no_omit(IDefaultConfigureForm, 'description')
    description = schema.Text(
        title=_(u'Description'),
        required=False,
    )

    form.omitted('date')
    form.no_omit(IDefaultConfigureForm, 'date')
    date = schema.Datetime(
        title=_(u'Date'),
        required=False,

app=makerequest(app)

from zope.component import getMultiAdapter
from zope.interface import Interface

try:
    portal = sys.argv[1]
except IndexError:
    portal = 'openplans'

app_view = getMultiAdapter((app, app.REQUEST), Interface,
                           name='manage_site.html')
app_view.unmakeSite()

portal = getattr(app, portal)
portal_view = getMultiAdapter((portal, portal.REQUEST), Interface,
                              name='manage_site.html')
portal_view.unmakeSite()

transaction.get().note('Unmade app and portal as local sites')
transaction.commit()

from .symlink import Symlink, follow_symlinks
from opennode.oms.model.model.events import IModelModifiedEvent, IModelCreatedEvent, IModelDeletedEvent


class ITokenized(Interface):
    def tokens():
        """Returns all tokens relevant for a model as a single string"""


class ITokenizer(Interface):
    def tokens():
        """Returns all tokens relevant for a model as a list of tokens"""


class ITagged(Interface):
    tags = schema.Set(title=u"Tags", required=False)


class ModelTokenizer(Adapter):
    implements(ITokenized)
    context(Model)

    def tokens(self):
        """Hackish way to quickly take all important tokens"""
        tokens = []
        if IDisplayName.providedBy(self.context):
            tokens.extend(IDisplayName(self.context).display_name().split('_'))

        if queryAdapter(self.context, ITagged):
            for tag in ITagged(self.context).tags:
                # hack, zope catalog treats ':' specially
                tokens.append(tag.replace(':', '_'))