Несколько баз данных

Приложение с несколькими подключениями выбирает подключение для каждого запроса: через using= в вызове или через маршрутизатор, который решает за все запросы модели, отдельно для чтения и записи.

using=

Подключение везде выбирается одинаково — одним именем using, которое принимает имя подключения ("replica") или клиент (клиент транзакции, клиент из autonomous(), маршрутизатора или Connections.get(connection_alias)):

У QuerySet: .using(alias_or_client) — копия запроса, закреплённая за этим подключением. С него начинается любой запрос, поэтому это покрывает create, get, get_or_create, update_or_create, filter, bulk_create, bulk_update, raw, count, ...: Book.objects.using("replica").filter(...), Book.objects.using(transaction).create(...).

У объекта: аргумент using= у save, delete, hard_delete, delete_preview, restore и refresh_from_db; у записей через связь (add, remove, set, clear, create, ...), в том числе «многие-ко-многим»; и у prefetch_related_objects().

В остальных местах: Transactions.atomic(using), on_commit(..., using=), on_rollback(..., using=), execute_sql(..., using=). Клиент автономной транзакции передаётся так же:

async with Transactions.autonomous() as connection:
    book = await Book.objects.using(connection).create(title="Notes")
    await book.tags.add(tag, using=connection)

Объект помнит своё подключение

Объект модели помнит имя подключения, из которого он загружен (включая объекты, построенные select_related()/prefetch_related(), stream(), union() и raw()), или в которое последний раз сохранён (save(), create(), get_or_create(), bulk_create()). Каждая следующая операция с ним без using= по умолчанию идёт туда: save(), delete(), restore(), refresh_from_db(), prefetch_related_objects(), ожидание прямой связи (await book.author) и связи объекта (author.books.all()/.filter()/.count()/.values(), add()/remove()/clear() у «многие-ко-многим»):

book = await Book.objects.all().using("replica").get(title="Notes")
author = await book.author          # прочитан из "replica", а не из "default"
books = await author.books.all()    # тоже "replica"

Подключение ищется по имени, когда выполняется запрос, поэтому внутри открытого atomic() на том же подключении используется транзакция. Явный using=/.using() всегда важнее, как и маршрутизатор, у которого есть мнение о запрашиваемой модели. Связанная модель, подключение по умолчанию которой отличается от подключения модели объекта (связь между разными базами), и связь, прочитанная из объекта модели, которую направляет маршрутизатор, сохраняют собственный выбор подключения. Queryset из Prefetch("books", Book.objects.filter(...)) читает из подключения родителя, как и простой prefetch_related("books"), если сам не выбрал подключение через .using().

Маршрутизаторы

Напишите обычный класс — наследовать ничего не нужно, hare вызывает его методы по имени — с одним или обоими методами, каждый из которых возвращает имя подключения (или None):

class ReportingRouter:
    def db_for_read(self, model: type[Model]) -> str | None:
        if model._meta.app == "reporting":
            return "reporting_replica"
        return None

    def db_for_write(self, model: type[Model]) -> str | None: ...

Любого из методов может не быть вовсе — hare пропускает маршрутизатор для действия, которого тот не реализует, а не выдаёт ошибку. Побеждает первый настроенный маршрутизатор, вернувший не None; если таких нет, используется обычное подключение модели по умолчанию. Возвращённое имя, которому не соответствует ни одно настроенное подключение, даёт ConfigurationError: маршрутизатор, который сработал, но назвал подключение с опечаткой или устаревшее, считается настоящей ошибкой настройки и не пропускается молча.

Чтение, от которого зависит запись, идёт через подключение db_for_write: проверка существования в get_or_create(), чтение с блокировкой в update_or_create() и select_for_update(). На реплике может ещё не быть строки, которая уже есть на основной базе (отставание репликации или строка, созданная раньше в ещё открытой транзакции), и промах там создал бы дубликат.

Регистрируются через Hare.init(routers=[...]) — списком путей через точку ("library.routers.ReportingRouter") или самих классов; каждый создаётся без аргументов. Каждый HareContext хранит свои маршрутизаторы в отдельном ConnectionRouter (context.router), который опрашивает их по очереди, — поэтому маршрутизаторы разных контекстов никогда не смешиваются.

Маршрутизатор выбирает, куда идут запросы, а не где лежат таблицы: makemigrations, migrate и drift работают с default_connection каждого приложения и маршрутизаторы не спрашивают. Поэтому маршрутизатор разводит запросы между подключениями с одними и теми же таблицами — основной базой и её репликами; модель, чья таблица лежит в другой базе, относится к приложению, у которого default_connection — эта база. Маршрутизатор, отправивший модель на подключение без её таблицы, получает там ошибку запроса (no such table).

Чтение своих записей

Реплика получает запись позже основной базы — запрос, который выдаёт книгу и сразу читает выдачу, может не найти её там. Поэтому, как только asyncio-задача (запрос) записала через подключение, каждое чтение модели, которую db_for_write направляет в это подключение (или которая пишется через своё default_connection, когда db_for_write возвращает None), идёт в него, а не в подключение из db_for_read, — до конца задачи. Другая задача — другой запрос — продолжает читать из реплики. Задача, запущенная из записавшей, тоже видит её записи; её собственные записи остаются при ней — запустившая её задача читает как прежде. Открытый Transactions.atomic("primary") — не запись: чтение до первой записи блока идёт туда, куда направляет db_for_read, — в реплику, а не в транзакцию. Транзакция, которая читает строки, чтобы их изменить, читает их через .using("primary") или внутри Routing.using_primary() — они берут открытую транзакцию.

Запись — то, что меняет строки: save(), delete() и restore() объекта, create(), update(), delete(), bulk_create() и bulk_update() у QuerySet, add(), remove(), clear() и set() связи «многие-ко-многим» — через подключение, выбранное маршрутизатором или названное в .using(). Вопрос, через какое подключение пишется модель (Model.get_connection(for_write=True)), ничего не пишет и чтений не меняет.

Настройка конфигурации read_your_writes_seconds заканчивает это через столько секунд после последней записи задачи, когда реплика успела догнать основную базу:

await Hare.init(
    config={**HARE_CONFIG, "routers": ["library.routers.PrimaryReplicaRouter"], "read_your_writes_seconds": 5}
)

Routing.using_primary() (блок with или async with) отправляет каждое чтение своего блока в подключение, через которое модель записывается, — писала задача или нет; это нужно, например, для проверки сразу после записи, сделанной в другом месте:

from hare.core.routing import Routing

async with Routing.using_primary():
    loan = await Loan.objects.get(pk=loan_id)

Routing.forget_writes() снова пускает чтения текущей задачи в реплики — для задачи, которая обслуживает несколько запросов подряд. .using() у QuerySet и транзакция выбирают подключение сами, и это их не касается. Без маршрутизаторов ничего из этого не работает.