Open a Zope FileStorage at a specific date and time, and copy that state to a new file. May give POSKeyError though.

gistfile1.txt

import ZODB.FileStorage
import logging
import sys
from ZODB.TimeStamp import TimeStamp

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('truncate')

if len(sys.argv) != 4:
    print("Usage: %s <source-filestorage> <target-filestorage> "
          "<UTC timestamp yyyy-mm-dd-[-hh[-mm[-ss]]]>" % sys.argv[0])
    sys.exit(1)

infile = sys.argv[1]
outfile = sys.argv[2]
timestamp = sys.argv[3]

date = [int(t) for t in timestamp.split('-')]
special_date = TimeStamp(*date)
date_id = repr(special_date)


def get_pos_from_filestorage(stop):
    # Reading a filestorage up to a certain transaction does not work with
    # an up to date index, so we do not read it.
    ZODB.FileStorage.FileStorage._restore_index = lambda x: None
    storage = ZODB.FileStorage.FileStorage(
        infile, read_only=1, stop=stop)
    pos = storage._file.tell()
    storage.close()
    return pos


def get_pos_from_fileiterator(stop):
    it = ZODB.FileStorage.FileIterator(infile, stop=stop)
    # Read all transactions from the iterator.  It will stop at the
    # point we want.
    tr = None
    count = 0
    for count, tr in enumerate(it):
        pass
    logger.info("read %d transactions", count)
    pos = it._pos
    # Do we need to ask the last transaction instead?
    # pos = tr._tend
    # When using the resulting filestorage, both give a POSKeyError in
    # a database I try it with...
    it.close()
    return pos


def humanize(size):
    # just add commas to indicate thousands
    h = [a for a in str(size)]
    h.reverse()
    res = []
    for index, i in enumerate(h):
        if index != 0 and index % 3 == 0:
            res.append(',')
        res.append(i)
    res.reverse()
    return ''.join(res)

logger.info("Reading %s ...", infile)
# Choose one of these functions:
# pos = get_pos_from_filestorage(date_id)
# This one seems cleanest:
pos = get_pos_from_fileiterator(date_id)

logger.info("Copying %s bytes from %s to %s ...",
            humanize(pos), infile, outfile)
if raw_input("Are you sure? (y/N) ").lower() != 'y':
    logger.info("Not copying.")
    sys.exit(0)
old = open(infile, 'rb')
new = open(outfile, 'wb')
# Inline alternative, losing all other data in the old file: old.truncate()
new.write(old.read(pos))
copied = new.tell()
new.close()
if copied != pos:
    # The help text says it is possible that you get less bytes, maybe
    # we need to handle that in a better way.
    logger.error("Expected %d bytes in %s, got %d",
        pos, outfile, copied)
else:
    logger.info("Successfully copied %d bytes.", copied)