Source code for ZODB.DemoStorage

##############################################################################
#
# Copyright (c) Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Demo ZODB storage

A demo storage supports demos by allowing a volatile changed database
to be layered over a base database.

The base storage must not change.

"""
from __future__ import print_function
import os
import random
import weakref
import tempfile
import ZODB.BaseStorage
import ZODB.blob
import ZODB.interfaces
import ZODB.MappingStorage
import ZODB.POSException
import ZODB.utils
import zope.interface

from .ConflictResolution import ConflictResolvingStorage
from .utils import load_current, maxtid

[docs]@zope.interface.implementer( ZODB.interfaces.IStorage, ZODB.interfaces.IStorageIteration, ) class DemoStorage(ConflictResolvingStorage): """A storage that stores changes against a read-only base database This storage was originally meant to support distribution of application demonstrations with populated read-only databases (on CDROM) and writable in-memory databases. Demo storages are extemely convenient for testing where setup of a base database can be shared by many tests. Demo storages are also handy for staging appplications where a read-only snapshot of a production database (often accomplished using a `beforestorage <https://pypi.org/project/zc.beforestorage/>`_) is combined with a changes database implemented with a :class:`~ZODB.FileStorage.FileStorage.FileStorage`. """
[docs] def __init__(self, name=None, base=None, changes=None, close_base_on_close=None, close_changes_on_close=None): """Create a demo storage :param str name: The storage name used by the :meth:`~ZODB.interfaces.IStorage.getName` and :meth:`~ZODB.interfaces.IStorage.sortKey` methods. :param object base: base storage :param object changes: changes storage :param bool close_base_on_close: A Flag indicating whether the base database should be closed when the demo storage is closed. :param bool close_changes_on_close: A Flag indicating whether the changes database should be closed when the demo storage is closed. If a base database isn't provided, a :class:`~ZODB.MappingStorage.MappingStorage` will be constructed and used. If ``close_base_on_close`` isn't specified, it will be ``True`` if a base database was provided and ``False`` otherwise. If a changes database isn't provided, a :class:`~ZODB.MappingStorage.MappingStorage` will be constructed and used and blob support will be provided using a temporary blob directory. If ``close_changes_on_close`` isn't specified, it will be ``True`` if a changes database was provided and ``False`` otherwise. """ if close_base_on_close is None: if base is None: base = ZODB.MappingStorage.MappingStorage() close_base_on_close = False else: close_base_on_close = True elif base is None: base = ZODB.MappingStorage.MappingStorage() self.base = base self.close_base_on_close = close_base_on_close if changes is None: self._temporary_changes = True changes = ZODB.MappingStorage.MappingStorage() zope.interface.alsoProvides(self, ZODB.interfaces.IBlobStorage) if close_changes_on_close is None: close_changes_on_close = False else: if ZODB.interfaces.IBlobStorage.providedBy(changes): zope.interface.alsoProvides(self, ZODB.interfaces.IBlobStorage) if close_changes_on_close is None: close_changes_on_close = True self.changes = changes self.close_changes_on_close = close_changes_on_close self._issued_oids = set() self._stored_oids = set() self._resolved = [] self._commit_lock = ZODB.utils.Lock() self._transaction = None if name is None: name = 'DemoStorage(%r, %r)' % (base.getName(), changes.getName()) self.__name__ = name self._copy_methods_from_changes(changes) self._next_oid = random.randint(1, 1<<62)
def _blobify(self): if (self._temporary_changes and isinstance(self.changes, ZODB.MappingStorage.MappingStorage) ): blob_dir = tempfile.mkdtemp('.demoblobs') _temporary_blobdirs[ weakref.ref(self, cleanup_temporary_blobdir) ] = blob_dir self.changes = ZODB.blob.BlobStorage(blob_dir, self.changes) self._copy_methods_from_changes(self.changes) return True def cleanup(self): self.base.cleanup() self.changes.cleanup() __opened = True def opened(self): return self.__opened def close(self): self.__opened = False if self.close_base_on_close: self.base.close() if self.close_changes_on_close: self.changes.close() def _copy_methods_from_changes(self, changes): for meth in ( '_lock', 'getSize', 'isReadOnly', 'sortKey', 'tpc_transaction', ): setattr(self, meth, getattr(changes, meth)) supportsUndo = getattr(changes, 'supportsUndo', None) if supportsUndo is not None and supportsUndo(): for meth in ('supportsUndo', 'undo', 'undoLog', 'undoInfo'): setattr(self, meth, getattr(changes, meth)) zope.interface.alsoProvides(self, ZODB.interfaces.IStorageUndoable) lastInvalidations = getattr(changes, 'lastInvalidations', None) if lastInvalidations is not None: self.lastInvalidations = lastInvalidations def getName(self): return self.__name__ __repr__ = getName def getTid(self, oid): try: return self.changes.getTid(oid) except ZODB.POSException.POSKeyError: return self.base.getTid(oid) def history(self, oid, size=1): try: r = self.changes.history(oid, size) except ZODB.POSException.POSKeyError: r = [] size -= len(r) if size: try: r += self.base.history(oid, size) except ZODB.POSException.POSKeyError: if not r: raise return r def iterator(self, start=None, end=None): for t in self.base.iterator(start, end): yield t for t in self.changes.iterator(start, end): yield t def lastTransaction(self): t = self.changes.lastTransaction() if t == ZODB.utils.z64: t = self.base.lastTransaction() return t def __len__(self): return len(self.changes) # still want load for old clients (e.g. zeo servers) load = load_current def loadBefore(self, oid, tid): try: result = self.changes.loadBefore(oid, tid) except ZODB.POSException.POSKeyError: # The oid isn't in the changes, so defer to base return self.base.loadBefore(oid, tid) if result is None: # The oid *was* in the changes, but there aren't any # earlier records. Maybe there are in the base. try: result = self.base.loadBefore(oid, tid) except ZODB.POSException.POSKeyError: # The oid isn't in the base, so None will be the right result pass else: if result and not result[-1]: # The oid is current in the base. We need to find # the end tid in the base by fining the first tid # in the changes. Unfortunately, there isn't an # api for this, so we have to walk back using # loadBefore. if tid == maxtid: # Special case: we were looking for the # current value. We won't find anything in # changes, so we're done. return result end_tid = maxtid t = self.changes.loadBefore(oid, end_tid) while t: end_tid = t[1] t = self.changes.loadBefore(oid, end_tid) result = result[:2] + ( end_tid if end_tid != maxtid else None, ) return result def loadBlob(self, oid, serial): try: return self.changes.loadBlob(oid, serial) except ZODB.POSException.POSKeyError: try: return self.base.loadBlob(oid, serial) except AttributeError: if not ZODB.interfaces.IBlobStorage.providedBy(self.base): raise ZODB.POSException.POSKeyError(oid, serial) raise except AttributeError: if self._blobify(): return self.loadBlob(oid, serial) raise def openCommittedBlobFile(self, oid, serial, blob=None): try: return self.changes.openCommittedBlobFile(oid, serial, blob) except ZODB.POSException.POSKeyError: try: return self.base.openCommittedBlobFile(oid, serial, blob) except AttributeError: if not ZODB.interfaces.IBlobStorage.providedBy(self.base): raise ZODB.POSException.POSKeyError(oid, serial) raise except AttributeError: if self._blobify(): return self.openCommittedBlobFile(oid, serial, blob) raise def loadSerial(self, oid, serial): try: return self.changes.loadSerial(oid, serial) except ZODB.POSException.POSKeyError: return self.base.loadSerial(oid, serial) def new_oid(self): with self._lock: while 1: oid = ZODB.utils.p64(self._next_oid ) if oid not in self._issued_oids: try: load_current(self.changes, oid) except ZODB.POSException.POSKeyError: try: load_current(self.base, oid) except ZODB.POSException.POSKeyError: self._next_oid += 1 self._issued_oids.add(oid) return oid self._next_oid = random.randint(1, 1<<62) def pack(self, t, referencesf, gc=None): if gc is None: if self._temporary_changes: return self.changes.pack(t, referencesf) elif self._temporary_changes: return self.changes.pack(t, referencesf, gc=gc) elif gc: raise TypeError( "Garbage collection isn't supported" " when there is a base storage.") try: self.changes.pack(t, referencesf, gc=False) except TypeError as v: if 'gc' in str(v): pass # The gc arg isn't supported. Don't pack raise
[docs] def pop(self): """Close the changes database and return the base. """ self.changes.close() return self.base
[docs] def push(self, changes=None): """Create a new demo storage using the storage as a base. The given changes are used as the changes for the returned storage and ``False`` is passed as ``close_base_on_close``. """ return self.__class__(base=self, changes=changes, close_base_on_close=False)
def store(self, oid, serial, data, version, transaction): assert version=='', "versions aren't supported" if transaction is not self._transaction: raise ZODB.POSException.StorageTransactionError(self, transaction) # Since the OID is being used, we don't have to keep up with it any # more. Save it now so we can forget it later. :) self._stored_oids.add(oid) # See if we already have changes for this oid try: old = load_current(self, oid)[1] except ZODB.POSException.POSKeyError: old = serial if old != serial: rdata = self.tryToResolveConflict(oid, old, serial, data) self.changes.store(oid, old, rdata, '', transaction) self._resolved.append(oid) else: self.changes.store(oid, serial, data, '', transaction) def storeBlob(self, oid, oldserial, data, blobfilename, version, transaction): assert version=='', "versions aren't supported" if transaction is not self._transaction: raise ZODB.POSException.StorageTransactionError(self, transaction) # Since the OID is being used, we don't have to keep up with it any # more. Save it now so we can forget it later. :) self._stored_oids.add(oid) try: self.changes.storeBlob( oid, oldserial, data, blobfilename, '', transaction) except AttributeError: if not self._blobify(): raise self.changes.storeBlob( oid, oldserial, data, blobfilename, '', transaction) checkCurrentSerialInTransaction = ( ZODB.BaseStorage.checkCurrentSerialInTransaction) def temporaryDirectory(self): try: return self.changes.temporaryDirectory() except AttributeError: if self._blobify(): return self.changes.temporaryDirectory() raise def tpc_abort(self, transaction): with self._lock: if transaction is not self._transaction: return self._stored_oids = set() self._transaction = None self.changes.tpc_abort(transaction) self._commit_lock.release() def tpc_begin(self, transaction, *a, **k): with self._lock: # The tid argument exists to support testing. if transaction is self._transaction: raise ZODB.POSException.StorageTransactionError( "Duplicate tpc_begin calls for same transaction") self._commit_lock.acquire() with self._lock: self.changes.tpc_begin(transaction, *a, **k) self._transaction = transaction self._stored_oids = set() del self._resolved[:] def tpc_vote(self, *a, **k): if self.changes.tpc_vote(*a, **k): raise ZODB.POSException.StorageTransactionError( "Unexpected resolved conflicts") return self._resolved def tpc_finish(self, transaction, func = lambda tid: None): with self._lock: if (transaction is not self._transaction): raise ZODB.POSException.StorageTransactionError( "tpc_finish called with wrong transaction") self._issued_oids.difference_update(self._stored_oids) self._stored_oids = set() self._transaction = None tid = self.changes.tpc_finish(transaction, func) self._commit_lock.release() return tid
_temporary_blobdirs = {} def cleanup_temporary_blobdir( ref, _temporary_blobdirs=_temporary_blobdirs, # Make sure it stays around ): blob_dir = _temporary_blobdirs.pop(ref, None) if blob_dir and os.path.exists(blob_dir): ZODB.blob.remove_committed_dir(blob_dir)