ZODB & Other Data Stores

[app:main]
use = egg:bookshelf_shelves
pyramid.reload_templates = true
zodbconn.uri = file://%(here)s/Data.fs?connection_cache_size=20000
retry.attempts = 3

[pshell]
setup = bookshelf_shelves.pshell.setup

import transaction
from persistent import Persistent
from persistent.list import PersistentList
from ZODB import DB


class Review(Persistent):
    def __init__(self, text):
        self.text = text
        self.tags = []                      # plain list: not tracked


db = DB(None)                               # None selects an in-memory storage
tm1 = transaction.TransactionManager()
tm2 = transaction.TransactionManager()
conn1, conn2 = db.open(tm1), db.open(tm2)

conn1.root()['r'] = Review('Great')
tm1.commit()
tm2.begin()

review = conn1.root()['r']
review.tags.append('classic')               # in-place change of a plain list
print('changed after append:', review._p_changed)
tm1.commit()
tm2.begin()
print('seen by the other connection:', conn2.root()['r'].tags)

review.tags = review.tags + ['sci-fi']      # fix 1: assign, so the attribute is set
print('changed after assignment:', review._p_changed)
tm1.commit()
tm2.begin()
print('seen by the other connection:', conn2.root()['r'].tags)

review.tags.append('epic')                  # fix 2: flag the owner by hand
review._p_changed = True
tm1.commit()
review.tags = PersistentList(review.tags)   # fix 3: use a persistent container from now on
review.tags.append('series')                # tracked on its own
tm1.commit()
tm2.begin()
print('final:', list(conn2.root()['r'].tags))

changed after append: False
seen by the other connection: []
changed after assignment: True
seen by the other connection: ['classic', 'sci-fi']
final: ['classic', 'sci-fi', 'epic', 'series']

import re

from persistent import Persistent
from persistent.list import PersistentList
from persistent.mapping import PersistentMapping
from pyramid.authorization import Allow, Everyone
from pyramid.config import Configurator
from pyramid.httpexceptions import HTTPBadRequest, HTTPConflict, HTTPSeeOther
from pyramid_zodbconn import get_connection


class Container(PersistentMapping):
    """A persistent, location-aware mapping: one node of the resource tree."""

    __name__ = None
    __parent__ = None


class Library(Container):
    pass


class Shelves(Container):
    pass


class Shelf(Container):
    def __init__(self, owner):
        super().__init__()
        self.owner = owner

    @property
    def __acl__(self):
        return [
            (Allow, Everyone, 'view'),
            (Allow, f'user:{self.owner}', ('add', 'edit')),
        ]


class Book(Persistent):
    __name__ = None
    __parent__ = None

    def __init__(self, title, author):
        self.title = title
        self.author = author
        self.tags = PersistentList()

    def __json__(self, request):
        return {'title': self.title, 'author': self.author,
                'tags': list(self.tags), 'url': request.resource_url(self)}


def add_child(container, name, child):
    child.__name__ = name
    child.__parent__ = container
    container[name] = child
    return child


def appmaker(zodb_root):
    """Create the application root on first use, as the wiki tutorial does."""
    if 'library' not in zodb_root:
        library = Library()
        add_child(library, 'shelves', Shelves())
        zodb_root['library'] = library
    return zodb_root['library']


def root_factory(request):
    return appmaker(get_connection(request).root())

NAME = re.compile(r'[a-z0-9-]{1,40}\Z')
RESERVED = {'add_shelf', 'add_book', 'read'}


def clean_name(request, key):
    name = request.POST.get(key, '')
    if not NAME.match(name) or name in RESERVED:
        raise HTTPBadRequest(f'invalid {key}')
    return name


def library_view(context, request):
    return {'shelves': request.resource_url(context['shelves'])}


def shelves_view(context, request):
    return {'shelves': sorted(context)}


def add_shelf(context, request):
    name = clean_name(request, 'name')
    if name in context:
        raise HTTPConflict('shelf exists')
    shelf = add_child(context, name, Shelf(owner=clean_name(request, 'owner')))
    return HTTPSeeOther(request.resource_url(shelf))


def shelf_view(context, request):
    return {'owner': context.owner, 'books': sorted(context)}


def add_book(context, request):
    name = clean_name(request, 'name')
    if name in context:
        raise HTTPConflict('book exists')
    book = add_child(context, name,
                     Book(request.POST['title'], request.POST['author']))
    return HTTPSeeOther(request.resource_url(book))


def book_view(context, request):
    return context


def make_app(settings, extra=None):
    with Configurator(settings=settings) as config:
        config.include('pyramid_tm')
        config.include('pyramid_retry')
        config.include('pyramid_zodbconn')
        config.set_root_factory(root_factory)
        config.add_view(library_view, context=Library, renderer='json')
        config.add_view(shelves_view, context=Shelves, renderer='json')
        config.add_view(add_shelf, context=Shelves, name='add_shelf',
                        request_method='POST')
        config.add_view(shelf_view, context=Shelf, renderer='json')
        config.add_view(add_book, context=Shelf, name='add_book',
                        request_method='POST')
        config.add_view(book_view, context=Book, renderer='json')
        if extra:
            config.include(extra)
    return config.make_wsgi_app()

from webtest import TestApp

app = make_app({'zodbconn.uri': 'memory://', 'retry.attempts': '3'})
client = TestApp(app)

print(client.post('/shelves/add_shelf', {'name': 'ada', 'owner': 'ada'}).location)
print(client.post('/shelves/ada/add_book', {'name': 'dune', 'title': 'Dune',
                                           'author': 'Frank Herbert'}).location)
print(client.get('/shelves/ada/dune').json)
print(client.get('/shelves/ada').json)
print(client.post('/shelves/add_shelf', {'name': 'add_shelf', 'owner': 'x'},
                  status=400).status)
print(client.post('/shelves/add_shelf', {'name': 'ada', 'owner': 'x'},
                  status=409).status)
client.get('/shelves/nobody', status=404)

http://localhost/shelves/ada/
http://localhost/shelves/ada/dune/
{'title': 'Dune', 'author': 'Frank Herbert', 'tags': [], 'url': 'http://localhost/shelves/ada/dune/'}
{'owner': 'ada', 'books': ['dune']}
400 Bad Request
409 Conflict

from pyramid.authorization import ACLHelper
from pyramid.scripting import prepare

with prepare(registry=app.registry) as env:
    shelf = env['root']['shelves']['ada']
    helper = ACLHelper()
    print(bool(helper.permits(shelf, ['user:ada'], 'edit')))
    print(bool(helper.permits(shelf, ['user:bob'], 'edit')))
    print(bool(helper.permits(shelf['dune'], [Everyone, 'user:bob'], 'view')))
    print(shelf['dune'].__parent__ is shelf)

import pyramid_zodbconn as pz

seen = []


def on_closing(event):
    seen.append((event.request.path, event.conn.getTransferCounts()))


logdir = tempfile.TemporaryDirectory()
logfile = os.path.join(logdir.name, 'transfers.csv')
settings = {
    'zodbconn.uri': 'memory://',
    'zodbconn.uri.sessions': 'memory://?database_name=sessions',
    'zodbconn.transferlog': logfile,
}
logged_app = make_app(settings, extra=lambda c: c.add_subscriber(on_closing, pz.ZODBConnectionWillClose))
TestApp(logged_app).get('/')
print(seen)
print(open(logfile).read().strip().split(', ')[1:2])

with prepare(registry=logged_app.registry) as env:
    main_conn = pz.get_connection(env['request'])
    sessions_conn = pz.get_connection(env['request'], 'sessions')
    print(repr(main_conn.db().database_name), repr(sessions_conn.db().database_name))

import pymongo
import redis
from bson import ObjectId
from pyramid.renderers import JSON


def includeme(config):
    settings = config.get_settings()
    registry = config.registry
    registry.redis = redis.Redis.from_url(settings['redis.url'], decode_responses=True)
    registry.mongo = pymongo.MongoClient(settings['mongo.url'])
    db_name = settings.get('mongo.db', 'bookshelf')

    config.add_request_method(lambda request: request.registry.redis, 'redis', reify=True)
    config.add_request_method(lambda request: request.registry.mongo[db_name], 'mongo_db', reify=True)
    config.add_request_method(after_commit, 'after_commit')
    config.add_request_method(redis_pipeline, 'redis_pipeline', reify=True)

    json_renderer = JSON()
    json_renderer.add_adapter(ObjectId, lambda value, request: str(value))
    config.add_renderer('json', json_renderer)
    config.add_view(read_book, context=Book, name='read', request_method='POST', renderer='json')
    config.add_view(read_and_fail, context=Book, name='read_fail', request_method='POST')
    config.add_view(events_view, context=Shelf, name='events', renderer='json')


def after_commit(request, func):
    """Run func only once the ZODB transaction has committed."""
    def hook(success):
        if success:
            func()
    request.tm.get().addAfterCommitHook(hook)


def redis_pipeline(request):
    pipe = request.redis.pipeline()
    request.add_finished_callback(lambda req: pipe.reset())
    request.after_commit(pipe.execute)
    return pipe


def read_book(context, request):
    key = f'reads:{context.__parent__.__name__}:{context.__name__}'
    request.redis_pipeline.incr(key)
    request.after_commit(lambda: request.mongo_db.events.insert_one(
        {'type': 'read', 'shelf': context.__parent__.__name__, 'book': context.__name__}))
    context.last_read = 'today'             # the transactional part, stored in ZODB
    return {'queued': key}


def read_and_fail(context, request):
    request.redis_pipeline.incr('reads:never')
    raise ValueError('boom')


def events_view(context, request):
    return list(request.mongo_db.events.find({'shelf': context.__name__}))

from unittest import mock

import fakeredis
import mongomock

with mock.patch('redis.Redis.from_url', fakeredis.FakeRedis.from_url), \
        mock.patch('pymongo.MongoClient', mongomock.MongoClient):
    stores_app = make_app({'zodbconn.uri': 'memory://',
                           'redis.url': 'redis://localhost:6379/0',
                           'mongo.url': 'mongodb://localhost:27017'}, extra=includeme)
stores = TestApp(stores_app)
stores.post('/shelves/add_shelf', {'name': 'ada', 'owner': 'ada'})
stores.post('/shelves/ada/add_book', {'name': 'dune', 'title': 'Dune', 'author': 'Frank Herbert'})
print(stores.post('/shelves/ada/dune/read').json)
stores.post('/shelves/ada/dune/read')
try:
    stores.post('/shelves/ada/dune/read_fail')
except ValueError as exc:
    print('raised', exc)
print(stores_app.registry.redis.get('reads:ada:dune'), stores_app.registry.redis.get('reads:never'))
print([e['type'] for e in stores.get('/shelves/ada/events').json])

{'queued': 'reads:ada:dune'}
raised boom
2 None
['read', 'read']