Dynamic Asynchronous Database Router with Instance Hints

from typing import Type, Optional
from django.db import models
 
class AsyncReplicaRouter:
    def db_for_read(self, model: Type[models.Model], **hints) -> str:
        if getattr(hints.get("instance"), "_use_primary", False):
            return "default"
        return "replica_async"
 
    def db_for_write(self, model: Type[models.Model], **hints) -> str:
        return "default"
 
    def allow_relation(self, obj1: models.Model, obj2: models.Model, **hints) -> Optional[bool]:
        db_set = {"default", "replica_async"}
        if obj1._state.db in db_set and obj2._state.db in db_set:
            return True
        return None