I'm a big fan of redis. It enabled me to scale my tornado apps and support realtime updates. Tornado recommends running behind nginx and bringing up multiple instances of your tornado app.
Now clients connect to an app and yet broadcasts need to go to all the clients of all the apps. Redis steps in and provides a pub/sub (publish and subscribe) pattern so that each app subscribes to a redis channel and the writing app calls publish. This works well on Heroku or AWS, where you do not know who or how many other servers there are. It does introduce a single point of failure.
So, what about Postgres? My persistence is via SQLAlchemy. I have occasionally switched from MySQL to Postgres and others to see if there was an advantage. MySQL had a memcache innodb interface that allowed you to update your cache within an innodb transaction at the same time that you updated your base data. Sadly, the people at MySQL have decided to depreciate the extension. So, I had a look at Postgres and their own connection classes, a look the other side of beyond SQLAlchemy. And there I found psycopg. The first thing I found of interest was the row_factory argument to the cursor constructor. You could think of this as a lite-alchemy. Your query result can be mapped to python tuples, namedtuple, dataclasses. So how is this useful?
@dataclass
class User:
email: str
name: str
id: int
prefs: Dict = field(default_factory=dict)
async def get_user(user_id: int) -> User:
async with async_cursor(row_factory=class_row(User)) as cursor:
rows = await cursor.execute(
'SELECT email, name, id, prefs FROM users',
{'id': user_id},
)
return await rows.fetchone()
Although we like to think of our classes an orthogonal, just like tables in SQL, that is only part of the story. We have views in SQL — in fact the very definition of SQL is that everything is a table — the result of every query is a table, a new table, a derived table. Let's take the concept of a user table.
In Postgres you would write:
CREATE TABLE user (
id SERIAL NOT NULL PRIMARY KEY,
name VARCHAR(255),
email VARCHAR(255) NOT NULL,
password VARCHAR(255),
prefs JSONB
);
In SQLAlchemy, we'd write:
user = Table("user", metadata,
Column("id", Integer, primary_key=True),
Column("name", String(255)),
Column("email", String(255), nullable=False),
Column("password", String(255)),
Column("prefs", JSON),
)
What use cases do we have for this user? Why, register, login, save_prefs and mailing_list. With each of these functions the user is represented by a different view.
The profile of a logged in user is:
@dataclass
class Profile:
id: int
name: str
email: str
prefs: Dict = field(default_factory=dict)
The dataclass for online_user:
@dataclass
class Online:
id: int
name: str
email: str
The dataclass for saved_prefs:
@dataclass
class SavedPrefs:
id: int
prefs: Dict = field(default_factory=dict)
The dataclass for a mailing_list:
@dataclass
class Recipient:
name: str
email: str
You get the picture. I can have the query make the view or I can use an orthogonal class User and rip the view from it. The most efficient is the former, but that is not the documented technique in SQLAlchemy. That is because it does the heavy lifting for inserts, updates and deletes of complete graphs. These views are for representation only.
We seem a long way from redis, and I apologise. Postgres has a Notify function and accompanying Listen. It allows for the pub/sub pattern. So we can use this instead of redis, since all our app are connected to the database for our persistence.
I created a channel to wrap this function:
class Channel:
"""
support for postgres Notify and Listen
"""
def __init__(self, name: str, listener: Callable) -> None:
self.name = name
self.listener = listener
self.background_tasks = []
def done_task(self, task):
self.background_tasks.remove(task)
Now the listener is the websocket class, which will broadcast to all its clients. The background_tasks is a weirdness introduced by asyncio, that the standard library will garbage collect tasks that do not have a strong reference!
def start_listening(self):
loop = asyncio.get_event_loop()
task = loop.create_task(self.listen(), name='listening')
task.add_done_callback(self.done_task)
self.background_tasks.append(task)
async def listen(self):
conn = None
loop = asyncio.get_event_loop()
try:
conn = await psycopg.AsyncConnection.connect(
os.environ['DATABASE_URL'],
autocommit=True
)
await conn.execute(f'LISTEN {self.name}'.encode())
async for item in conn.notifies():
if item.payload == 'stop':
break
else:
notify = loop.create_task(
self.listener(item.channel, item.payload, item.pid)
)
notify.add_done_callback(self.done_task)
self.background_tasks.append(notify)
except asyncio.CancelledError:
pass
except Exception as ex:
logging.exception(ex)
finally:
if conn:
await conn.close()
And then a utility to relay outside of an update:
async def relay(self, payload: str | bytes):
async with context.async_cursor() as cursor:
message = f'NOTIFY {self.name}, {quote(payload)}'
LOGGER.debug('relaying: %s', message)
await cursor.execute(message.encode('utf-8'))
await cursor.connection.commit()
or in SQLAlchemy:
async def sa_relay(self, payload: str | bytes):
async with context.async_session_factory() as session:
stmt = sa.text(f'Notify {self.name}, {quote(payload)}')
await session.execute(stmt)
await session.commit()
I'm not good enough to know how to get SQLAlchemy and psycopg to let through a statement like this. Bind parameters don't work and so I borrowed quote from psycopg.sql and build the statement with string formatting. It works and does the right thing even with quotation marks in the payload. Of course, the payload should be trusted - you wrote it.
You'll notice the explicit commit. For failing to commit will fail to send the message — that is the beauty — it's within a transaction, unlike redis! Let's just say without the need for two-phase commits.
So, now we have the best of both worlds! We have a light weight mapping of rows, full SQLAlchemy for inserts, updates and deletes, and pub/sub without the need for another process all within a transaction!
I hope this has been helpful.