Force queries in a context to execute in the named database

[rockisch]

Hey, I actually opened a ticket about this a while ago:

Allow 'Database Router' to override default `connection`

Yeah, that wouldn’t be that helpful xD It would just change the current proxy to fetch a different connection when getting a cursor. As it stands, connection is defined as a ConnectionProxy initialized with alias="default": class ConnectionProxy: """Proxy for accessing a connection object's attributes.""" def __init__(self, connections, alias): self.__dict__['_connections'] = connections self.__dict__['_alias'] = alias def __getattr__(self, item): return…

We actually recently deployed a ‘monkey-patch’ to do this kind of stuff in our project:

https://gist.github.com/rockisch/96425b6aaa406dcf4fb86df92715fe1e

If you’re only interested on overriding ORM queries, you could probably still it down to overriding the database router part:

import contextvars
from contextlib import contextmanager


_dbname_var = contextvars.ContextVar("dbname")
_readonly_var = contextvars.ContextVar("readonly")


@contextmanager
def db_override(dbname, readonly=False):
    dbname_token = _dbname_var.set(dbname)
    readonly_token = _readonly_var.set(readonly)
    try:
        yield
    finally:
        _dbname_var.reset(dbname_token)
        _readonly_var.reset(readonly_token)


class OverrideRouter:
    def db_for_read(self, *args, **kwargs):
        if dbname := _dbname_var.get(None):
            return dbname

    def db_for_write(self, *args, **kwargs):
        if dbname := _dbname_var.get(None):
            if _readonly_var.get(False):
                from django.db import InterfaceError

                raise InterfaceError(f"current db context does not allows writes to '{dbname}'")

            return dbname

However, any usage of transaction.atomic, connection.cursor or a lot of Django extensions usage might end up using the default DB (which is why I think routers right now are not that useful).