How to delete data from a RethinkDB database using its changefeed
<p>I'm working on a 'controller' for a database which continuously accrues data, but only uses recent data, defined as less than 3 days old. As soon as data becomes more than 3 days old, I'd like to dump it to a JSON file and remove it from the database.</p>

<p>To simulate this, I've done the following. The 'controller' program <code>rethinkdb_monitor.py</code> is</p>

<pre><code>import json
import rethinkdb as r
import pytz
from datetime import datetime, timedelta

# The database and table are assumed to have been previously created
database_name = "sensor_db"
table_name = "sensor_data"

port_offset = 1         # To avoid interference of this testing program with the main program, all ports are initialized at an offset of 1 from the default ports using "rethinkdb --port_offset 1" at the command line.
conn = r.connect("localhost", 28015 + port_offset)

current_time = datetime.utcnow().replace(tzinfo=pytz.utc)   # Current time include timezone (assumed UTC)
retention_period = timedelta(days=3)                        # Period of time during which data is retained on the main server
expiry_time = current_time - retention_period               # Age of data which is removed from the main server

data_to_archive = r.db(database_name).table(table_name).filter(r.row['timestamp'] &lt; expiry_time)
output_file = "archived_sensor_data.json"

with open(output_file, 'a') as f:
    for change in data_to_archive.changes().run(conn, time_format="raw"):        # The time_format="raw" option is passed to prevent a "RqlTzinfo object is not JSON serializable" error when dumping
        print change
        json.dump(change['new_val'], f)             # Since the main database we are reading from is append-only, the 'old_val' of the change is always None and we are interested in the 'new_val' only
        f.write("\n")                               # Separate entries by a new line
</code></pre>

<p>Prior to running this program, I started up RethinkDB using</p>

<pre><code>rethinkdb --port_offset 1
</code></pre>

<p>at the command line, and used the web interface at <code>localhost:8081</code> to create a database called <code>sensor_db</code> with a table called <code>sensor_data</code> (see below).</p>

<p><a href="https://i.stack.imgur.com/HaAPR.png" rel="nofollow"><img src="https://i.stack.imgur.com/HaAPR.png" alt="enter image description here"></a></p>

<p>Once <code>rethinkdb_monitor.py</code> is running and waiting for changes, I run a script <code>rethinkdb_add_data.py</code> which generates synthetic data:</p>

<pre><code>import random
import faker
from datetime import datetime, timedelta
import pytz
import rethinkdb as r

class RandomData(object):
    def __init__(self, seed=None):
        self._seed = seed
        self._random = random.Random()
        self._random.seed(seed)
        self.fake = faker.Faker()
        self.fake.random.seed(seed)

    def __getattr__(self, x):
        return getattr(self._random, x)

    def name(self):
        return self.fake.name()

    def datetime(self, start=None, end=None):
        if start is None:
            start = datetime(2000, 1, 1, tzinfo=pytz.utc)  # Jan 1st 2000
        if end is None:
            end = datetime.utcnow().replace(tzinfo=pytz.utc)

        if isinstance(end, datetime):
            dt = end - start
        elif isinstance(end, timedelta):
            dt = end
        assert isinstance(dt, timedelta)

        random_dt = timedelta(microseconds=self._random.randrange(int(dt.total_seconds() * (10 ** 6))))
        return start + random_dt

# Rethinkdb has been started at a port offset of 1 using the "--port_offset 1" argument.
port_offset = 1
conn = r.connect("localhost", 28015 + port_offset).repl()

rd = RandomData(seed=0)         # Instantiate and seed a random data generator

# The database and table have been previously created (e.g. through the web interface at localhost:8081)
database_name = "sensor_db"
table_name = "sensor_data"

# Generate random data with timestamps uniformly distributed over the past 6 days
random_data_time_interval = timedelta(days=6)
start_random_data = datetime.utcnow().replace(tzinfo=pytz.utc) - random_data_time_interval

for _ in range(5):
    entry = {"name": rd.name(), "timestamp": rd.datetime(start=start_random_data)}
    r.db(database_name).table(table_name).insert(entry).run()
</code></pre>

<p>After interrupting <code>rethinkdb_monitor.py</code> with Cntrl+C, the <code>archived_sensor_data.json</code> file contains the data to be archived:</p>

<pre><code>{"timestamp": {"timezone": "+00:00", "$reql_type$": "TIME", "epoch_time": 1475963599.347}, "id": "be2b5fd7-28df-48ee-b744-99856643265a", "name": "Elizabeth Woods"}
{"timestamp": {"timezone": "+00:00", "$reql_type$": "TIME", "epoch_time": 1475879797.486}, "id": "36d69236-f710-481b-82b6-4a62a1aae36c", "name": "Susan Wagner"}
</code></pre>

<p>What I am still struggling with, however, is how to subsequently remove this data from the DB. The command syntax of <code>delete</code> seems to be such that it can be called on a table or selection, but the <code>change</code> obtained through the changefeed is simply a dictionary.</p>

<p>How could I use the changefeed to continuously delete data from the database?</p>

<p>I used the fact that each <code>change</code> contains the ID of the corresponding document in the database, and created a selection using <code>get</code> with this ID:</p>

<pre><code>with open(output_file, 'a') as f:
    for change in data_to_archive.changes().run(conn, time_format="raw"):        # The time_format="raw" option is passed to prevent a "RqlTzinfo object is not JSON serializable" error when dumping
        print change
        if change['new_val'] is not None:               # If the change is not a deletion
            json.dump(change['new_val'], f)             # Since the main database we are reading from is append-only, the 'old_val' of the change is always None and we are interested in the 'new_val' only
            f.write("\n")                               # Separate entries by a new line
            ID_to_delete = change['new_val']['id']                # Get the ID of the data to be deleted from the database
            r.db(database_name).table(table_name).get(ID_to_delete).delete().run(conn)
</code></pre>

<p>The deletions will themselves be registered as changes, but I've used the <code>if change['new_val'] is not None</code> statement to filter these out.</p>

python rethinkdb