/usr/lib/python2.7/dist-packages/ZODB/fstools.py is in python-zodb 1:3.9.7-5.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | ##############################################################################
#
# Copyright (c) 2001, 2002 Zope Foundation 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
#
##############################################################################
"""Tools for using FileStorage data files.
TODO: This module needs tests.
Caution: This file needs to be kept in sync with FileStorage.py.
"""
import cPickle
import struct
from ZODB.FileStorage.format import TRANS_HDR, DATA_HDR, TRANS_HDR_LEN
from ZODB.FileStorage.format import DATA_HDR_LEN
from ZODB.utils import u64
from persistent.TimeStamp import TimeStamp
class TxnHeader:
"""Object representing a transaction record header.
Attribute Position Value
--------- -------- -----
tid 0- 8 transaction id
length 8-16 length of entire transaction record - 8
status 16-17 status of transaction (' ', 'u', 'p'?)
user_len 17-19 length of user field (pack code H)
descr_len 19-21 length of description field (pack code H)
ext_len 21-23 length of extensions (pack code H)
"""
def __init__(self, file, pos):
self._file = file
self._pos = pos
self._read_header()
def _read_header(self):
self._file.seek(self._pos)
self._hdr = self._file.read(TRANS_HDR_LEN)
(self.tid, self.length, self.status, self.user_len, self.descr_len,
self.ext_len) = struct.unpack(TRANS_HDR, self._hdr)
def read_meta(self):
"""Load user, descr, and ext attributes."""
self.user = ""
self.descr = ""
self.ext = {}
if not (self.user_len or self.descr_len or self.ext_len):
return
self._file.seek(self._pos + TRANS_HDR_LEN)
if self.user_len:
self.user = self._file.read(self.user_len)
if self.descr_len:
self.descr = self._file.read(self.descr_len)
if self.ext_len:
self._ext = self._file.read(self.ext_len)
self.ext = cPickle.loads(self._ext)
def get_data_offset(self):
return (self._pos + TRANS_HDR_LEN + self.user_len + self.descr_len
+ self.ext_len)
def get_timestamp(self):
return TimeStamp(self.tid)
def get_raw_data(self):
data_off = self.get_data_offset()
data_len = self.length - (data_off - self._pos)
self._file.seek(data_off)
return self._file.read(data_len)
def next_txn(self):
off = self._pos + self.length + 8
self._file.seek(off)
s = self._file.read(8)
if not s:
return None
return TxnHeader(self._file, off)
def prev_txn(self):
if self._pos == 4:
return None
self._file.seek(self._pos - 8)
tlen = u64(self._file.read(8))
return TxnHeader(self._file, self._pos - (tlen + 8))
class DataHeader:
"""Object representing a data record header.
Attribute Position Value
--------- -------- -----
oid 0- 8 object id
serial 8-16 object serial numver
prev_rec_pos 16-24 position of previous data record for object
txn_pos 24-32 position of txn header
version_len 32-34 length of version (always 0)
data_len 34-42 length of data
"""
def __init__(self, file, pos):
self._file = file
self._pos = pos
self._read_header()
def _read_header(self):
self._file.seek(self._pos)
self._hdr = self._file.read(DATA_HDR_LEN)
# always read the longer header, just in case
(self.oid, self.serial, prev_rec_pos, txn_pos, vlen, data_len
) = struct.unpack(DATA_HDR, self._hdr[:DATA_HDR_LEN])
assert not vlen
self.prev_rec_pos = u64(prev_rec_pos)
self.txn_pos = u64(txn_pos)
self.data_len = u64(data_len)
def next_offset(self):
"""Return offset of next record."""
off = self._pos + self.data_len
off += DATA_HDR_LEN
if self.data_len == 0:
off += 8 # backpointer
return off
def prev_txn(f):
"""Return transaction located before current file position."""
f.seek(-8, 1)
tlen = u64(f.read(8)) + 8
return TxnHeader(f, f.tell() - tlen)
|