ZODB: find oids of objects referencing a specific oid

zodb_refmap.py

"""
Usage
=====

First create a refmap of your ZODB:

>>> r = build_refmap('/path/to/Data.fs')

Now you can find out which objects reference an object you know the oid of:

>>> oid = '\x00\x00\x00\x00\x00uB\xf1'
>>> backrefs(oid, r)
['\x00\x00\x00\x00\x00uC\x1f']

As computing the refmap can consume some time, it can be stored on hard disk as
JSON data and re-read from there:

>>> write_refmap(r)
>>> r = read_refmap()

To find out the object behind the gotten oid, get it from the ZODB:

>>> import ZODB
>>> db = ZODB.DB('/path/to/Data.fs', read_only=1)
>>> c = db.open()
>>> obj = c.get('\x00\x00\x00\x00\x00uC\x1f')


To load a pure json from the storage use:

>>> fs = FileStorage('/path/to/Data.fs', read_only=1)
>>> fs.load('\x00\x00\x00\x00\x00uC\x1f')

"""

from ZODB.FileStorage import FileStorage
from ZODB.serialize import referencesf
from pprint import pprint
from pprint import pprint as pp  # noqa
import ZODB
import json


def build_refmap(filename):
    """Build a refmap from a filestorage.

    Look in every record of every transaction.
    Build a dict of oid -> set(oids referen)
    """
    refmap = {}
    fs = FileStorage(filename, read_only=1)
    fsi = fs.iterator()
    for txn in fsi:
        for rec in txn:
            pickle, revid = fs.load(rec.oid, rec.version)
            for ref in referencesf(pickle):
                r = refmap.setdefault(ref, set())
                r.add(rec.oid)
    return refmap


def write_refmap(refmap, name='refmap.json'):
    with open(name, 'wb') as f:
        unicode_refmap = {
            k.decode('latin-1'): [v.decode('latin-1') for v in values]
            for k, values in refmap.items()}
        json.dump(unicode_refmap, f)


def read_refmap(name='refmap.json'):
    with open(name, 'r') as f:
        unicode_refmap = json.load(f)
        return {k.encode('latin-1'): [v.encode('latin-1') for v in values]
                for k, values in unicode_refmap.items()}


def backrefs(target, refmap):
    """Return a list of oids in the refmap who reference target.
    """
    return refmap.get(target, set())


def obj_path(target, refmap):
    """For a target oid find the path of objects that refer to it.
       break if we reach no more references or find a cycle
    """
    path = [target]
    additionals = []
    while True:
        target = path[-1:].pop()
        brefs = backrefs(target, refmap)
        if not brefs:
            break
        bref = brefs[0]
        if bref in path:
            print('cyclic', bref)
            break
        if len(brefs) == 1:
            path.append(bref)
            print(bref)
            continue
        additionals.append((target, brefs[1:]))
        print(bref, brefs[1:])
        path.append(bref)
    return (path, additionals)

known = []

def step(refmap, conn, known, oid):
    """Step trough a database and keep track of the already seen oids."""
    if oid in known:
        print('cycle')
        return
    else:
        known.append(oid)
    brefs = backrefs(oid, refmap)
    pprint(brefs)
    pprint([conn.get(ref) for ref in brefs])

This looks useful - can this be added to one of the zodbverify / zodbupgrade / zombi-whatever tools?

@dwt zodbverify now has the ability to look at the references of an oid.

@icemac: 👍

Added capability to store/load the refmap to/from hard disk + fix the data structure so accessing a OID no longer requires to search the whole refmap.