18153960591

Committed 01 Oct 2025 06:50AM UTC coverage: 83.781% (-0.03%) from 83.811%

Build # 18153960591

Build Type

Pull #415

github

Committed by

web-flow

Commit Message

Update docs/articles/old-guide/convert_zodb_guide.py

Co-authored-by: Michael Howitz <icemac@gmx.net>

Pull Request Pull Request #415: Apply the latest zope.meta templates

Run Details

2441 of 3542 branches covered (68.92%)

193 of 257 new or added lines in 48 files covered. (75.1%)

12 existing lines in 6 files now uncovered.

13353 of 15938 relevant lines covered (83.78%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.26

/src/ZODB/DemoStorage.py

##############################################################################
#
# 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.

"""

import os
import random
import tempfile
import weakref

import zope.interface

import ZODB.BaseStorage
import ZODB.blob
import ZODB.interfaces
import ZODB.MappingStorage
import ZODB.POSException
import ZODB.utils

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


@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`.
    """

    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 = f'DemoStorage({base.getName()!r}, {changes.getName()!r})'
        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):
        yield from self.base.iterator(start, end)
        yield from self.changes.iterator(start, end)

    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

    def pop(self):
        """Close the changes database and return the base.
        """
        self.changes.close()
        return self.base

    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)

1	##############################################################################
2	#
3	# Copyright (c) Zope Corporation and Contributors.
4	# All Rights Reserved.
5	#
6	# This software is subject to the provisions of the Zope Public License,
7	# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
8	# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
9	# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
10	# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
11	# FOR A PARTICULAR PURPOSE
12	#
13	##############################################################################
14	"""Demo ZODB storage
15
16	A demo storage supports demos by allowing a volatile changed database
17	to be layered over a base database.
18
19	The base storage must not change.
20
21	"""
22
23	import os	1✔
24	import random	1✔
25	import tempfile	1✔
26	import weakref	1✔
27
28	import zope.interface	1✔
29
30	import ZODB.BaseStorage	1✔
31	import ZODB.blob	1✔
32	import ZODB.interfaces	1✔
33	import ZODB.MappingStorage	1✔
34	import ZODB.POSException	1✔
35	import ZODB.utils	1✔
36
37	from .ConflictResolution import ConflictResolvingStorage	1✔
38	from .utils import load_current	1✔
39	from .utils import maxtid	1✔
40
41
42	@zope.interface.implementer(	1✔
43	ZODB.interfaces.IStorage,
44	ZODB.interfaces.IStorageIteration,
45	)
46	class DemoStorage(ConflictResolvingStorage):	1✔
47	"""A storage that stores changes against a read-only base database
48
49	This storage was originally meant to support distribution of
50	application demonstrations with populated read-only databases (on
51	CDROM) and writable in-memory databases.
52
53	Demo storages are extemely convenient for testing where setup of a
54	base database can be shared by many tests.
55
56	Demo storages are also handy for staging appplications where a
57	read-only snapshot of a production database (often accomplished
58	using a `beforestorage
59	<https://pypi.org/project/zc.beforestorage/>`_) is combined
60	with a changes database implemented with a
61	:class:`~ZODB.FileStorage.FileStorage.FileStorage`.
62	"""
63
64	def __init__(self, name=None, base=None, changes=None,	1✔
65	close_base_on_close=None, close_changes_on_close=None):
66	"""Create a demo storage
67
68	:param str name: The storage name used by the
69	:meth:`~ZODB.interfaces.IStorage.getName` and
70	:meth:`~ZODB.interfaces.IStorage.sortKey` methods.
71	:param object base: base storage
72	:param object changes: changes storage
73	:param bool close_base_on_close: A Flag indicating whether the base
74	database should be closed when the demo storage is closed.
75	:param bool close_changes_on_close: A Flag indicating whether the
76	changes database should be closed when the demo storage is closed.
77
78	If a base database isn't provided, a
79	:class:`~ZODB.MappingStorage.MappingStorage` will be
80	constructed and used.
81
82	If ``close_base_on_close`` isn't specified, it will be ``True`` if
83	a base database was provided and ``False`` otherwise.
84
85	If a changes database isn't provided, a
86	:class:`~ZODB.MappingStorage.MappingStorage` will be
87	constructed and used and blob support will be provided using a
88	temporary blob directory.
89
90	If ``close_changes_on_close`` isn't specified, it will be ``True`` if
91	a changes database was provided and ``False`` otherwise.
92	"""
93
94	if close_base_on_close is None:	1✔
95	if base is None:	1✔
96	base = ZODB.MappingStorage.MappingStorage()	1✔
97	close_base_on_close = False	1✔
98	else:
99	close_base_on_close = True	1✔
100	elif base is None:	1!
101	base = ZODB.MappingStorage.MappingStorage()	×
102
103	self.base = base	1✔
104	self.close_base_on_close = close_base_on_close	1✔
105
106	if changes is None:	1✔
107	self._temporary_changes = True	1✔
108	changes = ZODB.MappingStorage.MappingStorage()	1✔
109	zope.interface.alsoProvides(self, ZODB.interfaces.IBlobStorage)	1✔
110	if close_changes_on_close is None:	1!
111	close_changes_on_close = False	1✔
112	else:
113	if ZODB.interfaces.IBlobStorage.providedBy(changes):	1✔
114	zope.interface.alsoProvides(self, ZODB.interfaces.IBlobStorage)	1✔
115	if close_changes_on_close is None:	1✔
116	close_changes_on_close = True	1✔
117
118	self.changes = changes	1✔
119	self.close_changes_on_close = close_changes_on_close	1✔
120
121	self._issued_oids = set()	1✔
122	self._stored_oids = set()	1✔
123	self._resolved = []	1✔
124
125	self._commit_lock = ZODB.utils.Lock()	1✔
126	self._transaction = None	1✔
127
128	if name is None:	1✔
129	name = f'DemoStorage({base.getName()!r}, {changes.getName()!r})'	1✔
130	self.__name__ = name	1✔
131
132	self._copy_methods_from_changes(changes)	1✔
133
134	self._next_oid = random.randint(1, 1 << 62)	1✔
135
136	def _blobify(self):	1✔
137	if (self._temporary_changes and	1!
138	isinstance(self.changes, ZODB.MappingStorage.MappingStorage)):
139	blob_dir = tempfile.mkdtemp('.demoblobs')	1✔
140	_temporary_blobdirs[	1✔
141	weakref.ref(self, cleanup_temporary_blobdir)
142	] = blob_dir
143	self.changes = ZODB.blob.BlobStorage(blob_dir, self.changes)	1✔
144	self._copy_methods_from_changes(self.changes)	1✔
145	return True	1✔
146
147	def cleanup(self):	1✔
148	self.base.cleanup()	1✔
149	self.changes.cleanup()	1✔
150
151	__opened = True	1✔
152
153	def opened(self):	1✔
154	return self.__opened	1✔
155
156	def close(self):	1✔
157	self.__opened = False	1✔
158	if self.close_base_on_close:	1✔
159	self.base.close()	1✔
160	if self.close_changes_on_close:	1✔
161	self.changes.close()	1✔
162
163	def _copy_methods_from_changes(self, changes):	1✔
164	for meth in (	1✔
165	'_lock',
166	'getSize', 'isReadOnly',
167	'sortKey', 'tpc_transaction',
168	):
169	setattr(self, meth, getattr(changes, meth))	1✔
170
171	supportsUndo = getattr(changes, 'supportsUndo', None)	1✔
172	if supportsUndo is not None and supportsUndo():	1✔
173	for meth in ('supportsUndo', 'undo', 'undoLog', 'undoInfo'):	1✔
174	setattr(self, meth, getattr(changes, meth))	1✔
175	zope.interface.alsoProvides(self, ZODB.interfaces.IStorageUndoable)	1✔
176
177	lastInvalidations = getattr(changes, 'lastInvalidations', None)	1✔
178	if lastInvalidations is not None:	1✔
179	self.lastInvalidations = lastInvalidations	1✔
180
181	def getName(self):	1✔
182	return self.__name__	1✔
183	__repr__ = getName	1✔
184
185	def getTid(self, oid):	1✔
186	try:	1✔
187	return self.changes.getTid(oid)	1✔
188	except ZODB.POSException.POSKeyError:	1✔
189	return self.base.getTid(oid)	1✔
190
191	def history(self, oid, size=1):	1✔
192	try:	1✔
193	r = self.changes.history(oid, size)	1✔
194	except ZODB.POSException.POSKeyError:	1✔
195	r = []	1✔
196	size -= len(r)	1✔
197	if size:	1✔
198	try:	1✔
199	r += self.base.history(oid, size)	1✔
200	except ZODB.POSException.POSKeyError:	1✔
201	if not r:	1!
202	raise	1✔
203	return r	1✔
204
205	def iterator(self, start=None, end=None):	1✔
206	yield from self.base.iterator(start, end)	1✔
207	yield from self.changes.iterator(start, end)	1✔
208
209	def lastTransaction(self):	1✔
210	t = self.changes.lastTransaction()	1✔
211	if t == ZODB.utils.z64:	1✔
212	t = self.base.lastTransaction()	1✔
213	return t	1✔
214
215	def __len__(self):	1✔
216	return len(self.changes)	1✔
217
218	# still want load for old clients (e.g. zeo servers)
219	load = load_current	1✔
220
221	def loadBefore(self, oid, tid):	1✔
222	try:	1✔
223	result = self.changes.loadBefore(oid, tid)	1✔
224	except ZODB.POSException.POSKeyError:	1✔
225	# The oid isn't in the changes, so defer to base
226	return self.base.loadBefore(oid, tid)	1✔
227
228	if result is None:	1✔
229	# The oid was in the changes, but there aren't any
230	# earlier records. Maybe there are in the base.
231	try:	1✔
232	result = self.base.loadBefore(oid, tid)	1✔
233	except ZODB.POSException.POSKeyError:	1✔
234	# The oid isn't in the base, so None will be the right result
235	pass	1✔
236	else:
237	if result and not result[-1]:	1!
238	# The oid is current in the base. We need to find
239	# the end tid in the base by fining the first tid
240	# in the changes. Unfortunately, there isn't an
241	# api for this, so we have to walk back using
242	# loadBefore.
243
244	if tid == maxtid:	1!
245	# Special case: we were looking for the
246	# current value. We won't find anything in
247	# changes, so we're done.
248	return result	×
249
250	end_tid = maxtid	1✔
251	t = self.changes.loadBefore(oid, end_tid)	1✔
252	while t:	1✔
253	end_tid = t[1]	1✔
254	t = self.changes.loadBefore(oid, end_tid)	1✔
255	result = result[:2] + (	1✔
256	end_tid if end_tid != maxtid else None,
257	)
258
259	return result	1✔
260
261	def loadBlob(self, oid, serial):	1✔
262	try:	1✔
263	return self.changes.loadBlob(oid, serial)	1✔
264	except ZODB.POSException.POSKeyError:	1✔
265	try:	1✔
266	return self.base.loadBlob(oid, serial)	1✔
267	except AttributeError:	1✔
268	if not ZODB.interfaces.IBlobStorage.providedBy(self.base):	1!
269	raise ZODB.POSException.POSKeyError(oid, serial)	1✔
270	raise	×
271	except AttributeError:	1✔
272	if self._blobify():	1!
273	return self.loadBlob(oid, serial)	1✔
274	raise	×
275
276	def openCommittedBlobFile(self, oid, serial, blob=None):	1✔
277	try:	1✔
278	return self.changes.openCommittedBlobFile(oid, serial, blob)	1✔
279	except ZODB.POSException.POSKeyError:	1✔
280	try:	1✔
281	return self.base.openCommittedBlobFile(oid, serial, blob)	1✔
282	except AttributeError:	1✔
283	if not ZODB.interfaces.IBlobStorage.providedBy(self.base):	1!
284	raise ZODB.POSException.POSKeyError(oid, serial)	1✔
285	raise	×
286	except AttributeError:	×
287	if self._blobify():	×
288	return self.openCommittedBlobFile(oid, serial, blob)	×
289	raise	×
290
291	def loadSerial(self, oid, serial):	1✔
292	try:	1✔
293	return self.changes.loadSerial(oid, serial)	1✔
294	except ZODB.POSException.POSKeyError:	1✔
295	return self.base.loadSerial(oid, serial)	1✔
296
297	def new_oid(self):	1✔
298	with self._lock:	1✔
299	while 1:	1✔
300	oid = ZODB.utils.p64(self._next_oid)	1✔
301	if oid not in self._issued_oids:	1✔
302	try:	1✔
303	load_current(self.changes, oid)	1✔
304	except ZODB.POSException.POSKeyError:	1✔
305	try:	1✔
306	load_current(self.base, oid)	1✔
307	except ZODB.POSException.POSKeyError:	1✔
308	self._next_oid += 1	1✔
309	self._issued_oids.add(oid)	1✔
310	return oid	1✔
311
312	self._next_oid = random.randint(1, 1 << 62)	1✔
313
314	def pack(self, t, referencesf, gc=None):	1✔
315	if gc is None:	1✔
316	if self._temporary_changes:	1!
317	return self.changes.pack(t, referencesf)	1✔
318	elif self._temporary_changes:	1!
319	return self.changes.pack(t, referencesf, gc=gc)	1✔
320	elif gc:	×
321	raise TypeError(	×
322	"Garbage collection isn't supported"
323	" when there is a base storage.")
324
325	try:	×
326	self.changes.pack(t, referencesf, gc=False)	×
327	except TypeError as v:	×
328	if 'gc' in str(v):	×
UNCOV 329	pass # The gc arg isn't supported. Don't pack	×
330	raise	×
331
332	def pop(self):	1✔
333	"""Close the changes database and return the base.
334	"""
335	self.changes.close()	1✔
336	return self.base	1✔
337
338	def push(self, changes=None):	1✔
339	"""Create a new demo storage using the storage as a base.
340
341	The given changes are used as the changes for the returned
342	storage and ``False`` is passed as ``close_base_on_close``.
343	"""
344	return self.__class__(base=self, changes=changes,	1✔
345	close_base_on_close=False)
346
347	def store(self, oid, serial, data, version, transaction):	1✔
348	assert version == '', "versions aren't supported"	1✔
349	if transaction is not self._transaction:	1✔
350	raise ZODB.POSException.StorageTransactionError(self, transaction)	1✔
351
352	# Since the OID is being used, we don't have to keep up with it any
353	# more. Save it now so we can forget it later. :)
354	self._stored_oids.add(oid)	1✔
355
356	# See if we already have changes for this oid
357	try:	1✔
358	old = load_current(self, oid)[1]	1✔
359	except ZODB.POSException.POSKeyError:	1✔
360	old = serial	1✔
361
362	if old != serial:	1✔
363	rdata = self.tryToResolveConflict(oid, old, serial, data)	1✔
364	self.changes.store(oid, old, rdata, '', transaction)	×
365	self._resolved.append(oid)	×
366	else:
367	self.changes.store(oid, serial, data, '', transaction)	1✔
368
369	def storeBlob(self, oid, oldserial, data, blobfilename, version,	1✔
370	transaction):
371	assert version == '', "versions aren't supported"	1✔
372	if transaction is not self._transaction:	1!
373	raise ZODB.POSException.StorageTransactionError(self, transaction)	×
374
375	# Since the OID is being used, we don't have to keep up with it any
376	# more. Save it now so we can forget it later. :)
377	self._stored_oids.add(oid)	1✔
378
379	try:	1✔
380	self.changes.storeBlob(	1✔
381	oid, oldserial, data, blobfilename, '', transaction)
382	except AttributeError:	1✔
383	if not self._blobify():	1!
384	raise	×
385	self.changes.storeBlob(	1✔
386	oid, oldserial, data, blobfilename, '', transaction)
387
388	checkCurrentSerialInTransaction = (	1✔
389	ZODB.BaseStorage.checkCurrentSerialInTransaction)
390
391	def temporaryDirectory(self):	1✔
392	try:	1✔
393	return self.changes.temporaryDirectory()	1✔
394	except AttributeError:	×
395	if self._blobify():	×
396	return self.changes.temporaryDirectory()	×
397	raise	×
398
399	def tpc_abort(self, transaction):	1✔
400	with self._lock:	1✔
401	if transaction is not self._transaction:	1✔
402	return	1✔
403	self._stored_oids = set()	1✔
404	self._transaction = None	1✔
405	self.changes.tpc_abort(transaction)	1✔
406	self._commit_lock.release()	1✔
407
408	def tpc_begin(self, transaction, a, *k):	1✔
409	with self._lock:	1✔
410	# The tid argument exists to support testing.
411	if transaction is self._transaction:	1✔
412	raise ZODB.POSException.StorageTransactionError(	1✔
413	"Duplicate tpc_begin calls for same transaction")
414
415	self._commit_lock.acquire()	1✔
416
417	with self._lock:	1✔
418	self.changes.tpc_begin(transaction, a, *k)	1✔
419	self._transaction = transaction	1✔
420	self._stored_oids = set()	1✔
421	del self._resolved[:]	1✔
422
423	def tpc_vote(self, a, *k):	1✔
424	if self.changes.tpc_vote(a, *k):	1!
425	raise ZODB.POSException.StorageTransactionError(	×
426	"Unexpected resolved conflicts")
427	return self._resolved	1✔
428
429	def tpc_finish(self, transaction, func=lambda tid: None):	1✔
430	with self._lock:	1✔
431	if (transaction is not self._transaction):	1✔
432	raise ZODB.POSException.StorageTransactionError(	1✔
433	"tpc_finish called with wrong transaction")
434	self._issued_oids.difference_update(self._stored_oids)	1✔
435	self._stored_oids = set()	1✔
436	self._transaction = None	1✔
437	tid = self.changes.tpc_finish(transaction, func)	1✔
438	self._commit_lock.release()	1✔
439	return tid	1✔
440
441
442	_temporary_blobdirs = {}	1✔
443
444
445	def cleanup_temporary_blobdir(	1✔
446	ref,
447	_temporary_blobdirs=_temporary_blobdirs, # Make sure it stays around
448	):
449	blob_dir = _temporary_blobdirs.pop(ref, None)	1✔
450	if blob_dir and os.path.exists(blob_dir):	1✔
451	ZODB.blob.remove_committed_dir(blob_dir)	1✔

zopefoundation / ZODB / 18153960591

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous