Skip to content

Instantly share code, notes, and snippets.

Last active April 16, 2023 15:12
Show Gist options
  • Save adamsvoboda/76d353d3d272d6c828af2319b10ae7fd to your computer and use it in GitHub Desktop.
Save adamsvoboda/76d353d3d272d6c828af2319b10ae7fd to your computer and use it in GitHub Desktop.
asynchronous rethinkdb change feeds with django channels
import asyncio
from rethinkdb import r, RqlRuntimeError
from channels import Group
r.set_loop_type('asyncio') # RethinkDB's driver must use asyncio for the loop type.
def create_changefeed_monitor(feeds):
Creates a new asyncio event loop and starts the parallel execution of attach_changefeed
for each table we want to monitor. Easily extended to accept whatever arguements you need.
Once your routing is setup, this function could be called like:
Channel('create-changefeed-monitor').send({'tables': ['table1', 'table2']})
loop = asyncio.new_event_loop()
tasks = []
for table in feeds['tables']:
# RethinkDB Changefeeds
async def changefeed_attach(table):
Creates a new RethinkDB connection for each instance of the function, attaches to
the changefeed for the table, awaits for any changes to be emitted from the
AsyncioCursor returned by RethinkDB!
The python RethinkDB driver is not thread-safe, but I don't think creating new
connections for each changefeed monitor is a big deal. "RethinkDB doesn't create a
thread per connection, so once established, you can have tens of thousands of
connections with very little overhead."
It may be worth investigating a Queue.Queue() object and doing something like a
RethinkDB connection pool for projects that use a lot of RethinkDB.
conn = await r.connect('', 28015, db='your_database')
feed = await r.table(table).changes().run(conn)
while await feed.fetch_next():
change = await
# Do something with the change. I'm broadcasting it out to a group for testing.
Group('changefeed-' + table).send({
"type": "changefeed",
"table": table,
"old": change['old_val'],
"new": change['new_val'],
except RqlRuntimeError as e:
# TODO: Implement changefeed error recovery or task monitoring
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment