Topics

Following database changes

The newt.db.follow module provides an API for subscribing to database changes. Newt DB uses it for asynchronous updates, but it can be used for other applications too.

What the stream is for

• Creating data indexes, such as Elasticsearch indexes.
• Creating alternate representations, such as relational models to support other applications.
• Monitoring or analytics of data changes.

The updates iterator

You can get an iterator of changes by calling the updates() function:

import newt.db.follow
import pickle

for batch in newt.db.follow.updates(dsn):
    for tid, zoid, data in batch:
        print(zoid, pickle.loads(data).__name__)

The updates iterator returns batches to make batch processing possible while still processing data as soon as possible. Batches are as large as possible, limited loosely by a target batch size, with the constraint that transactions are not split between batches. Batches are themselves iterators.

The iterator can run over a range of transactions, or it can run indefinitely, returning new batches as new data are committed. One of the parameters is end_tid, an end transaction id for the iteration. If no end_tid is provided, the iterator will iterate forever, blocking when necessary to wait for new data to be committed.

Convert records to JSON

The data returned by the follower is a pickle, which is probably not very useful on its own. You can convert it to JSON using the Newt DB JSON conversion. Here is the example above, updated:

import newt.db.follow
import newt.db.jsonpickle

jsonifier = newt.db.jsonpickle.Jsonifier()
for batch in newt.db.follow.updates(dsn):
    for tid, zoid, data in batch:
        class_name, _, data = jsonifier((zoid, tid), data)
        if data is not None:
            print(zoid, class_name, data)

Jsonifiers take a label, used for logging errors, and data, and return a class name, a ghost pickle and object state as JSON data. The ghost pickle is generally only useful to Newt DB itself, so we ignore it here. If the JSON data returned is None, we skip processing the data.

The return value may be None if:

• the raw data was an empty string, in which case the database record deleted the object;
• the object class was one the JSON conversion skipped;
• there was an error converting the data.

Tracking progress

Often the data returned by the updates iterator is used to update some other data. Clients are often stopped and later restarted, and need to keep track of where they left off. The set_progress_tid() method saves progress for a client:

newt.db.follow.set_progress_tid(dsn, 'mypackage.mymodule', tid)

The first argument is a PostgreSQL database connection. The second argument is a client identifier, typically the dotted name of the client module. The third argument is the last transaction id that was processed.

Later, you can use get_progress_tid() to retrieve the saved transaction id:

start_tid = newt.db.follow.get_progress_tid(dsn, 'mypackage.mymodule')

You then pass the retrieved transaction identifier as the start_tid argument to updates().

Garbage collection

One complication in dealing with external data is garbage collection. When a Newt DB database is packed, records are removed without generating updates. Data that is removed from Newt DB when it is packed should be removed from external representations as well. The easiest way to do this is by splitting packing into three steps.

1. Identify the garbage

   Run zodbpack with the --prepack option. This tells zodbpack to stop after identifying garbage.

   zodbpack -d 1 --prepack CONFIG_FILE

2. Remove the external copies

   Call newt.db.follow.garbage() to get an iterator of object ids that will be deleted in the second phase of packing.

   import newt.db.follow
   
   for zoid in newt.db.follow.garbage(dsn):
       my_remove_external_data_function(zoid)

3. Finish the pack

   Run zodbpack with the --use-prepack-state option. This tells zodbpack to remove the garbage identified in the first step.

   zodbpack -d 1 --use-prepack-state CONFIG_FILE