django-rw-router 0.0.3

Django database read-write separation router with transaction-aware and consistent hash routing support

Django Read-Write Router

Django 读写分离路由器，支持事务感知与一致性哈希路由。

Features

• 基础读写分离：写走主库，读走从库
• 事务感知路由： 事务内读自动走主库（读己之写）
• 一致性哈希路由： 同一用户始终读同一从库（单调读）
• 请求上下文追踪： 中间件自动管理请求上下文
• 主库 QuerySet： 可强制指定查询走主库

Installation

pip install django-rw-router

Or with uv:

uv add django-rw-router

Quick Start

1. 配置数据库

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'USER': 'user',
        'PASSWORD': 'password',
        'HOST': 'primary.db.example.com',
        'PORT': '5432',
    },
    'readonly': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'USER': 'readonly_user',
        'PASSWORD': 'password',
        'HOST': 'replica.db.example.com',
        'PORT': '5432',
    },
}

2. 添加路由器

# settings.py
DATABASE_ROUTERS = [
    'django_rw_router.routers.TransactionPrimaryReplicaRouter',
]

3. 添加中间件（HashPrimaryReplicaRouter 必选）

# settings.py
MIDDLEWARE = [
    'django_rw_router.middleware.RequestContextMiddleware',
    # ... 其他中间件
]

Router Options

路由器	场景	一致性
PrimaryReplicaRouter	基础读写分离	最终一致
TransactionPrimaryReplicaRouter	写后读	Read Your Writes
HashPrimaryReplicaRouter	同一用户多次读	Monotonic Reads
PrimaryReplicaRouterWithCtx	追踪写操作上下文	-

PrimaryReplicaRouter

基础读写分离。写走主库，读在从库间随机分配。

DATABASE_ROUTERS = [
    'django_rw_router.routers.PrimaryReplicaRouter',
]

TransactionPrimaryReplicaRouter（推荐）

事务内读写均走主库，避免复制滞后导致的"读己之写"不可见。

DATABASE_ROUTERS = [
    'django_rw_router.routers.TransactionPrimaryReplicaRouter',
]

HashPrimaryReplicaRouter

基于 user_id 的一致性哈希，同一用户始终读同一从库，保证单调读。

DATABASE_ROUTERS = [
    'django_rw_router.routers.hash.HashPrimaryReplicaRouter',
]
# 需启用 RequestContextMiddleware 以注入 user_id

PrimaryReplicaRouterWithCtx

在基础路由器上增加写操作上下文追踪，供上层逻辑判断。

DATABASE_ROUTERS = [
    'django_rw_router.routers.PrimaryReplicaRouterWithCtx',
]

一致性场景 (DDIA)

场景	含义	推荐路由器
Read Your Writes	写入后立即读可见	TransactionPrimaryReplicaRouter
Monotonic Reads	同一用户不出现时光倒流	HashPrimaryReplicaRouter
Consistent Prefix	有序读取不出现乱序	HashPrimaryReplicaRouter
Eventual Consistency	接受复制滞后	PrimaryReplicaRouter

Configuration

数据库别名

# settings.py

# 读从库别名（单个或列表）
DJANGO_RW_ROUTER_READ_DBS = ['readonly', 'readonly2']

# 写主库别名
DJANGO_RW_ROUTER_WRITE_DB = 'default'

# 一致性哈希虚拟节点数（HashPrimaryReplicaRouter）
DJANGO_RW_ROUTER_HASH_VIRTUAL_NODES = 40

中间件配置

# settings.py

# 从 request 提取 user_id 的路径（支持嵌套，如 user.id）
DJANGO_RW_ROUTER_USER_ID_ATTR = 'user.id'

# 从 request 提取 request_id 的路径
DJANGO_RW_ROUTER_REQUEST_ID_ATTR = 'id'

QuerySet 配置

# settings.py

# 强制 PrimaryQuerySet 所有读走主库
DJANGO_RW_ROUTER_QUERYSET_USING_ENABLE = True

# 启用 @use_primary_db 装饰器
DJANGO_RW_ROUTER_METHOD_USING_ENABLE = True

Usage Examples

Basic Usage

from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=100)

# Writes go to 'default', reads go to 'readonly'
book = Book.objects.create(title="Django Guide")  # Write to primary
books = Book.objects.all()  # Read from replica

Transaction-Aware Reads

from django.db import transaction

with transaction.atomic():
    book = Book.objects.create(title="New Book")
    # This read goes to PRIMARY (not replica) because we're in a transaction
    fresh_book = Book.objects.get(id=book.id)

Using PrimaryManager for Critical Queries

from django.db import models
from django_rw_router.managers import PrimaryManager

class ImportantModel(models.Model):
    objects = PrimaryManager()

# When DJANGO_RW_ROUTER_QUERYSET_USING_ENABLE=True,
# all reads through this manager go to primary
obj = ImportantModel.objects.first()

Manual Database Selection

You can always override the router:

# Force read from primary
obj = Book.objects.using('default').first()

# Force write to replica (not recommended)
book.save(using='readonly')

How It Works

1. 写操作：始终路由到主库 (default)
2. 读操作：
   • 事务外：路由到随机从库
   • 事务内：路由到主库（避免读己之写不可见）
3. Hash 路由：user_id 哈希后稳定选中同一从库
4. 上下文：中间件在请求开始时注入、结束时清理

Testing

# uv
uv run pytest -v

# pip
pip install -e ".[dev]"
pytest -v

测试覆盖 context、routers、managers、middleware，以及 DDIA 四种一致性场景。

Requirements

• Python >= 3.8
• Django >= 3.2
• uhashring >= 2.1

License

MIT License

Contributing

欢迎提交 Pull Request。