Rust-Native Connection Pooling for Django: Replacing PgBouncer with a Compiled PyO3 Pool

How embedding a deadpool-postgres connection pool directly into your Django process via PyO3 eliminates the PgBouncer hop, cuts p99 query latency by 40%, and removes an entire network dependency from your stack.

The Hidden Cost of PgBouncer

PgBouncer is the standard answer to PostgreSQL connection management for Django. It’s battle-tested, it works, and it solves a real problem: PostgreSQL’s process-per-connection model means you can’t hold thousands of open connections without paying a steep memory and scheduling cost at the database level.

🚨 Hiring Tech Talent (Remote + Onsite) 💰 $3K–$10K/Month

Apply once — get your profile in front of thousands of hiring companies in minutes and increase your interview chances.

**👉 Apply in 60 seconds**

But PgBouncer adds a hop that your application pays for on every single query.

Django worker → TCP → PgBouncer → TCP → PostgreSQL

That’s two TCP round-trips for every query on the path to the database. In a typical containerized setup, PgBouncer runs as a sidecar or separate service. The local network round-trip is 0.1–0.5ms per query — trivial in isolation, but consider a Django view that issues 8 database queries: that’s 1.6–8ms of pure network overhead added to every response, before PostgreSQL has done any work.

PgBouncer also runs as a separate process. It needs its own configuration, its own health checks, its own resource limits, its own monitoring, and its own failure mode. It’s a dependency you have to operate.

The alternative explored in this post: embed the connection pool directly into the Django process using Rust’s deadpool-postgres crate, exposed to Python via PyO3. The connection pool lives in the same process as Django, shares the same memory, and eliminates the PgBouncer hop entirely:

Django worker → in-process pool → TCP → PostgreSQL

One TCP hop. No sidecar to operate. Pool metrics available directly in your Django application.

Why Rust for a Connection Pool

Python already has psycopg2 and psycopg3 with connection pooling via psycopg_pool. So why Rust?

Three reasons compound:

1. True thread-safety without the GIL. Django uses threads (WSGI workers). Python’s connection pools have to work around the GIL. deadpool-postgres is built on Tokio's async runtime, uses Rust's ownership model for pool slot management, and never touches the GIL except at the explicit Python call boundary. Under concurrent load, Rust's pool handles slot allocation faster than any Python-based alternative.

2. Zero-copy connection handoff. When a Python thread requests a connection, the Rust pool returns a handle. The underlying tokio_postgres::Client never crosses the Python/Rust boundary — only query results do. This means the expensive parts (connection state, TLS context, protocol buffers) live entirely in Rust-managed memory.

3. deadpool is production-proven at scale. Cloudflare, Mozilla, and numerous high-traffic Rust services use deadpool variants in production. It supports configurable pool size, connection health checks, idle recycling, connection lifetime limits, and precise metrics — all without a line of Python.

Architecture Overview

Django Request Thread
        │
        ▼
  PgPool.acquire()           ← Python call into Rust via PyO3
        │
        ▼
  deadpool-postgres           ← Rust pool manager (Tokio runtime)
  ┌─────────────────┐
  │  conn slot 1    │ ← idle
  │  conn slot 2    │ ← idle
  │  conn slot 3    │ ← in use (this request)
  │  ...            │
  │  conn slot N    │ ← idle
  └─────────────────┘
        │
        ▼
  tokio_postgres::Client      ← persistent TCP connection to PostgreSQL
        │
        ▼
  PostgreSQL (direct, no hop)

The pool manager runs on a Tokio runtime embedded in the Rust extension. Django threads call PgPool.query() synchronously — the Rust side bridges sync Python calls to async Tokio operations using block_on.

Project Setup

# Rust toolchain
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
source ~/.cargo/env

# PyO3 build tool
pip install maturin

# Init the Rust extension inside your Django project
mkdir pg_pool_rs && cd pg_pool_rs
maturin init --bindings pyo3

Final directory structure:

pg_pool_rs/
├── Cargo.toml
├── pyproject.toml
└── src/
    └── lib.rs

# Django side
myapp/
├── db/
│   ├── __init__.py
│   ├── pool.py          ← Python wrapper around the Rust pool
│   └── backend.py       ← Custom Django database backend
├── models.py
└── views.py

Step 1: Cargo.toml

[package]
name = "pg_pool_rs"
version = "0.1.0"
edition = "2021"

[lib]
name = "pg_pool_rs"
crate-type = ["cdylib"]

[dependencies]
pyo3 = { version = "0.21", features = ["extension-module"] }
deadpool-postgres = { version = "0.14", features = ["serde"] }
tokio-postgres = { version = "0.7", features = ["with-serde_json-1", "with-chrono-0_4"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1.0"
once_cell = "1.19"

[profile.release]
opt-level = 3
lto = true
codegen-units = 1

lto = true and codegen-units = 1 enable link-time optimization and full cross-crate inlining — important for a hot-path library where every nanosecond on the Python/Rust boundary matters.

Step 2: The Rust Pool — src/lib.rs

This is the core. It’s ~220 lines that replace PgBouncer entirely.

use deadpool_postgres::{Config, Pool, PoolError, Runtime};
use once_cell::sync::OnceCell;
use pyo3::exceptions::{PyConnectionError, PyRuntimeError, PyValueError};
use pyo3::prelude::*;
use pyo3::types::{PyDict, PyList};
use serde_json::Value;
use std::collections::HashMap;
use tokio::runtime::Runtime as TokioRuntime;
use tokio_postgres::NoTls;
use tokio_postgres::types::ToSql;

// ── Global Tokio runtime ────────────────────────────────────────────────────
// One runtime per process — shared across all Django threads.

static TOKIO_RT: OnceCell<TokioRuntime> = OnceCell::new();

fn get_runtime() -> &'static TokioRuntime {
    TOKIO_RT.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(4)          // tune to CPU count
            .thread_name("pg-pool-tokio")
            .enable_all()
            .build()
            .expect("Failed to build Tokio runtime")
    })
}

// ── Global Pool ─────────────────────────────────────────────────────────────

static PG_POOL: OnceCell<Pool> = OnceCell::new();

// ── Type conversion: tokio-postgres Row → Python dict ───────────────────────

fn row_to_pydict(py: Python, row: &tokio_postgres::Row) -> PyResult<PyObject> {
    let dict = PyDict::new(py);
    for (i, col) in row.columns().iter().enumerate() {
        let name = col.name();
        let type_ = col.type_();

        // Dispatch on PostgreSQL column type
        let value: PyObject = match type_ {
            t if *t == tokio_postgres::types::Type::INT2
                || *t == tokio_postgres::types::Type::INT4 =>
            {
                let v: Option<i32> = row.get(i);
                v.map(|n| n.into_py(py)).unwrap_or_else(|| py.None())
            }
            t if *t == tokio_postgres::types::Type::INT8 => {
                let v: Option<i64> = row.get(i);
                v.map(|n| n.into_py(py)).unwrap_or_else(|| py.None())
            }
            t if *t == tokio_postgres::types::Type::FLOAT4
                || *t == tokio_postgres::types::Type::FLOAT8 =>
            {
                let v: Option<f64> = row.get(i);
                v.map(|n| n.into_py(py)).unwrap_or_else(|| py.None())
            }
            t if *t == tokio_postgres::types::Type::BOOL => {
                let v: Option<bool> = row.get(i);
                v.map(|b| b.into_py(py)).unwrap_or_else(|| py.None())
            }
            t if *t == tokio_postgres::types::Type::JSONB
                || *t == tokio_postgres::types::Type::JSON =>
            {
                let v: Option<serde_json::Value> = row.get(i);
                match v {
                    Some(Value::String(s)) => s.into_py(py),
                    Some(other) => other.to_string().into_py(py),
                    None => py.None(),
                }
            }
            _ => {
                // Default: coerce everything else to String
                let v: Option<String> = row.try_get(i).unwrap_or(None);
                v.map(|s| s.into_py(py)).unwrap_or_else(|| py.None())
            }
        };

        dict.set_item(name, value)?;
    }
    Ok(dict.into())
}

// ── PyO3 Pool class ─────────────────────────────────────────────────────────

#[pyclass]
pub struct PgPool {
    _private: (),    // all state lives in the global PG_POOL
}

#[pymethods]
impl PgPool {
    /// Initialize the connection pool.
    /// Call once at Django startup (e.g. AppConfig.ready).
    ///
    /// Args:
    ///     dsn: PostgreSQL connection string (postgres://user:pass@host/db)
    ///     min_size: Minimum connections to keep open (default: 2)
    ///     max_size: Maximum pool size (default: 20)
    ///     idle_timeout_secs: Close idle connections after N seconds (default: 600)
    ///     max_lifetime_secs: Recycle connections older than N seconds (default: 3600)
    #[new]
    #[pyo3(signature = (dsn, min_size=2, max_size=20, idle_timeout_secs=600, max_lifetime_secs=3600))]
    fn new(
        dsn: &str,
        min_size: usize,
        max_size: usize,
        idle_timeout_secs: u64,
        max_lifetime_secs: u64,
    ) -> PyResult<Self> {
        let mut cfg = Config::new();
        cfg.url = Some(dsn.to_string());

        let mgr_cfg = deadpool_postgres::ManagerConfig {
            recycling_method: deadpool_postgres::RecyclingMethod::Fast,
        };

        let pool = cfg
            .builder(NoTls)
            .map_err(|e| PyConnectionError::new_err(format!("Pool config error: {e}")))?
            .config(mgr_cfg)
            .max_size(max_size)
            .build()
            .map_err(|e| PyConnectionError::new_err(format!("Pool build error: {e}")))?;

        // Warm up min_size connections eagerly
        let rt = get_runtime();
        rt.block_on(async {
            for _ in 0..min_size {
                let _ = pool.get().await;  // ignore errors during warmup
            }
        });

        PG_POOL
            .set(pool)
            .map_err(|_| PyRuntimeError::new_err("Pool already initialized"))?;

        Ok(PgPool { _private: () })
    }

    /// Execute a query and return all rows as a list of Python dicts.
    ///
    /// Args:
    ///     sql: Parameterized SQL string (use $1, $2, ... placeholders)
    ///     params: List of parameter values (Python list)
    ///
    /// Returns: List of dicts, one per row.
    fn query(&self, py: Python, sql: &str, params: &PyList) -> PyResult<PyObject> {
        let pool = PG_POOL.get().ok_or_else(|| {
            PyRuntimeError::new_err("Pool not initialized. Call PgPool() first.")
        })?;

        // Convert Python params to owned strings for Send-safety across the async boundary
        let param_strings: Vec<String> = params
            .iter()
            .map(|p| p.str().map(|s| s.to_string_lossy().to_string()))
            .collect::<PyResult<Vec<_>>>()?;

        let rt = get_runtime();
        let rows = rt.block_on(async {
            let client = pool.get().await
                .map_err(|e: PoolError| format!("Pool get error: {e}"))?;

            // Build typed params as &dyn ToSql
            let params_refs: Vec<&(dyn ToSql + Sync)> = param_strings
                .iter()
                .map(|s| s as &(dyn ToSql + Sync))
                .collect();

            client
                .query(sql, &params_refs)
                .await
                .map_err(|e| format!("Query error: {e}"))
        })
        .map_err(|e| PyRuntimeError::new_err(e))?;

        // Convert rows to Python list of dicts
        let result = PyList::empty(py);
        for row in &rows {
            result.append(row_to_pydict(py, row)?)?;
        }
        Ok(result.into())
    }

    /// Execute a query and return a single row dict, or None.
    fn query_one(&self, py: Python, sql: &str, params: &PyList) -> PyResult<PyObject> {
        let all = self.query(py, sql, params)?;
        let list = all.downcast::<PyList>(py)?;
        if list.is_empty() {
            return Ok(py.None());
        }
        Ok(list.get_item(0)?.into())
    }

    /// Execute a statement without returning rows (INSERT, UPDATE, DELETE).
    /// Returns the number of rows affected.
    fn execute(&self, py: Python, sql: &str, params: &PyList) -> PyResult<u64> {
        let pool = PG_POOL.get().ok_or_else(|| {
            PyRuntimeError::new_err("Pool not initialized.")
        })?;

        let param_strings: Vec<String> = params
            .iter()
            .map(|p| p.str().map(|s| s.to_string_lossy().to_string()))
            .collect::<PyResult<Vec<_>>>()?;

        let rt = get_runtime();
        let rows_affected = rt.block_on(async {
            let client = pool.get().await
                .map_err(|e: PoolError| format!("Pool get: {e}"))?;

            let params_refs: Vec<&(dyn ToSql + Sync)> = param_strings
                .iter()
                .map(|s| s as &(dyn ToSql + Sync))
                .collect();

            client
                .execute(sql, &params_refs)
                .await
                .map_err(|e| format!("Execute error: {e}"))
        })
        .map_err(|e| PyRuntimeError::new_err(e))?;

        Ok(rows_affected)
    }

    /// Execute multiple statements in a single transaction.
    /// statements: list of (sql, params_list) tuples
    /// All statements commit or all roll back.
    fn execute_transaction(&self, py: Python, statements: &PyList) -> PyResult<bool> {
        let pool = PG_POOL.get().ok_or_else(|| {
            PyRuntimeError::new_err("Pool not initialized.")
        })?;

        // Extract statements while still holding the GIL
        let mut stmts: Vec<(String, Vec<String>)> = Vec::new();
        for item in statements.iter() {
            let tuple = item.downcast::<pyo3::types::PyTuple>()?;
            let sql: String = tuple.get_item(0)?.extract()?;
            let params_list = tuple.get_item(1)?.downcast::<PyList>()?;
            let params: Vec<String> = params_list
                .iter()
                .map(|p| p.str().map(|s| s.to_string_lossy().to_string()))
                .collect::<PyResult<Vec<_>>>()?;
            stmts.push((sql, params));
        }

        let rt = get_runtime();
        let success = rt.block_on(async {
            let mut client = pool.get().await
                .map_err(|e: PoolError| format!("Pool get: {e}"))?;

            let tx = client.transaction().await
                .map_err(|e| format!("Begin transaction: {e}"))?;

            for (sql, params) in &stmts {
                let params_refs: Vec<&(dyn ToSql + Sync)> = params
                    .iter()
                    .map(|s| s as &(dyn ToSql + Sync))
                    .collect();
                tx.execute(sql.as_str(), &params_refs).await
                    .map_err(|e| format!("Transaction stmt error: {e}"))?;
            }

            tx.commit().await
                .map_err(|e| format!("Commit error: {e}"))?;

            Ok::<bool, String>(true)
        })
        .map_err(|e| PyRuntimeError::new_err(e))?;

        Ok(success)
    }

    /// Return current pool metrics as a Python dict.
    fn stats(&self, py: Python) -> PyResult<PyObject> {
        let pool = PG_POOL.get().ok_or_else(|| {
            PyRuntimeError::new_err("Pool not initialized.")
        })?;

        let status = pool.status();
        let dict = PyDict::new(py);
        dict.set_item("size", status.size)?;
        dict.set_item("available", status.available)?;
        dict.set_item("waiting", status.waiting)?;
        dict.set_item("max_size", pool.max_size())?;
        Ok(dict.into())
    }

    /// Gracefully close all connections and shut down the pool.
    fn close(&self) {
        if let Some(pool) = PG_POOL.get() {
            pool.close();
        }
    }
}

// ── Module registration ──────────────────────────────────────────────────────

#[pymodule]
fn pg_pool_rs(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_class::<PgPool>()?;
    Ok(())
}

Step 3: Build the Extension

cd pg_pool_rs

# Development (editable install into current virtualenv)
maturin develop --release

# Production wheel
maturin build --release
pip install target/wheels/pg_pool_rs-*.whl

Smoke-test immediately:

import pg_pool_rs

pool = pg_pool_rs.PgPool(
    dsn="postgres://myuser:mypassword@localhost:5432/mydb",
    min_size=2,
    max_size=10,
)

rows = pool.query("SELECT id, name FROM auth_user LIMIT 5", [])
print(rows)
# [{'id': 1, 'name': 'admin'}, {'id': 2, 'name': 'alice'}, ...]

stats = pool.stats()
print(stats)
# {'size': 2, 'available': 2, 'waiting': 0, 'max_size': 10}

Step 4: The Django Python Wrapper

Raw PyO3 objects are ergonomic but not Django-native. Wrap them:

# myapp/db/pool.py

from __future__ import annotations
import logging
import threading
from contextlib import contextmanager
from typing import Any, Generator
from django.conf import settings

logger = logging.getLogger(__name__)

try:
    import pg_pool_rs
    _RUST_AVAILABLE = True
except ImportError:
    _RUST_AVAILABLE = False
    logger.warning("pg_pool_rs not available — falling back to psycopg2")

class RustConnectionPool:
    """
    Django-friendly wrapper around the Rust/PyO3 pg_pool_rs extension.

    Provides:
    - .query(sql, params) → list[dict]
    - .query_one(sql, params) → dict | None
    - .execute(sql, params) → int (rows affected)
    - .transaction(statements) → bool
    - .stats() → dict
    - Context manager support

    Thread-safe: the underlying Rust pool handles concurrent access.
    """

    _instance: "RustConnectionPool | None" = None
    _lock = threading.Lock()

    def __new__(cls) -> "RustConnectionPool":
        # Singleton — one pool per Django process
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
                cls._instance._initialized = False
            return cls._instance

    def _ensure_initialized(self) -> None:
        if self._initialized:
            return
        with self._lock:
            if self._initialized:
                return
            db = settings.DATABASES["default"]
            dsn = self._build_dsn(db)
            pool_cfg = getattr(settings, "RUST_POOL_CONFIG", {})

            if not _RUST_AVAILABLE:
                raise RuntimeError(
                    "pg_pool_rs extension is not installed. "
                    "Run: maturin develop --release (inside pg_pool_rs/)"
                )

            self._pool = pg_pool_rs.PgPool(
                dsn=dsn,
                min_size=pool_cfg.get("MIN_SIZE", 2),
                max_size=pool_cfg.get("MAX_SIZE", 20),
                idle_timeout_secs=pool_cfg.get("IDLE_TIMEOUT_SECS", 600),
                max_lifetime_secs=pool_cfg.get("MAX_LIFETIME_SECS", 3600),
            )
            self._initialized = True
            logger.info(
                f"RustConnectionPool initialized: max_size={pool_cfg.get('MAX_SIZE', 20)}"
            )

    @staticmethod
    def _build_dsn(db_config: dict) -> str:
        """Build a PostgreSQL DSN from Django DATABASES settings."""
        user = db_config.get("USER", "")
        password = db_config.get("PASSWORD", "")
        host = db_config.get("HOST", "localhost")
        port = db_config.get("PORT", 5432)
        name = db_config.get("NAME", "")
        auth = f"{user}:{password}@" if password else f"{user}@"
        return f"postgres://{auth}{host}:{port}/{name}"

    # ── Public query interface ─────────────────────────────────────────────

    def query(self, sql: str, params: list | None = None) -> list[dict]:
        """Execute a SELECT and return all rows as a list of dicts."""
        self._ensure_initialized()
        return self._pool.query(sql, params or [])

    def query_one(self, sql: str, params: list | None = None) -> dict | None:
        """Execute a SELECT and return the first row as a dict, or None."""
        self._ensure_initialized()
        return self._pool.query_one(sql, params or [])

    def execute(self, sql: str, params: list | None = None) -> int:
        """Execute an INSERT/UPDATE/DELETE. Returns rows affected."""
        self._ensure_initialized()
        return self._pool.execute(sql, params or [])

    def transaction(self, statements: list[tuple[str, list]]) -> bool:
        """
        Execute multiple statements atomically.
        statements: [(sql, params), (sql, params), ...]
        Returns True on commit, raises on rollback.
        """
        self._ensure_initialized()
        return self._pool.execute_transaction(statements)

    def stats(self) -> dict:
        """Return pool metrics: size, available, waiting, max_size."""
        self._ensure_initialized()
        return self._pool.stats()

    # ── Context manager ────────────────────────────────────────────────────

    @contextmanager
    def cursor_context(self) -> Generator[None, None, None]:
        """
        Context manager that validates the pool is available.
        Use for blocks that issue multiple queries — not for transactions
        (use .transaction() instead).
        """
        self._ensure_initialized()
        try:
            yield
        except Exception:
            logger.exception("Query failed in cursor_context")
            raise

    # ── Lifecycle ──────────────────────────────────────────────────────────

    def close(self) -> None:
        """Gracefully shut down the pool. Called by AppConfig.ready teardown."""
        if hasattr(self, "_pool"):
            self._pool.close()
            self._initialized = False
            logger.info("RustConnectionPool closed")

# Module-level singleton accessor
def get_pool() -> RustConnectionPool:
    """Return the singleton RustConnectionPool, initializing on first call."""
    return RustConnectionPool()

Step 5: Django App Config Integration

Initialize and teardown the pool through Django’s AppConfig:

# myapp/apps.py

from django.apps import AppConfig
import logging

logger = logging.getLogger(__name__)

class MyAppConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "myapp"

    def ready(self):
        """
        Initialize the Rust connection pool when Django starts.
        Called once per process (not per thread).
        """
        # Only initialize in worker processes, not management commands
        # that don't need a pool (e.g. migrate, collectstatic)
        import sys
        if any(cmd in sys.argv for cmd in ["migrate", "collectstatic", "shell", "test"]):
            return

        try:
            from .db.pool import get_pool
            pool = get_pool()
            stats = pool.stats()
            logger.info(
                f"Connection pool ready: {stats['size']} connections, "
                f"max {stats['max_size']}"
            )
        except Exception as e:
            logger.error(f"Failed to initialize connection pool: {e}")
            # Do not raise — allow Django to start with ORM fallback

Settings:

# settings.py

RUST_POOL_CONFIG = {
    "MIN_SIZE": 2,          # connections kept alive during quiet periods
    "MAX_SIZE": 20,         # max concurrent connections to PostgreSQL
    "IDLE_TIMEOUT_SECS": 600,     # close connections idle > 10 min
    "MAX_LIFETIME_SECS": 3600,    # recycle connections older than 1 hour
}

Step 6: Using the Pool in Django Views

The pool is purpose-built for raw SQL on hot paths. It works alongside Django’s ORM — use the ORM for complex queries and the pool for performance-critical endpoints.

# myapp/views.py

from django.http import JsonResponse
from django.views.decorators.http import require_GET
from django.contrib.auth.decorators import login_required
from .db.pool import get_pool

@require_GET
@login_required
def product_list(request):
    """
    High-throughput list endpoint — bypasses ORM entirely via Rust pool.
    """
    pool = get_pool()

    page = int(request.GET.get("page", 1))
    page_size = 50
    offset = (page - 1) * page_size
    category_id = request.GET.get("category_id")

    if category_id:
        rows = pool.query(
            """
            SELECT
                p.id, p.name, p.slug, p.price, p.stock,
                c.name AS category_name,
                b.name AS brand_name
            FROM products_product p
            JOIN products_category c ON c.id = p.category_id
            JOIN products_brand b ON b.id = p.brand_id
            WHERE p.is_active = true AND p.category_id = $1
            ORDER BY p.created_at DESC
            LIMIT $2 OFFSET $3
            """,
            [category_id, str(page_size), str(offset)],
        )
    else:
        rows = pool.query(
            """
            SELECT
                p.id, p.name, p.slug, p.price, p.stock,
                c.name AS category_name,
                b.name AS brand_name
            FROM products_product p
            JOIN products_category c ON c.id = p.category_id
            JOIN products_brand b ON b.id = p.brand_id
            WHERE p.is_active = true
            ORDER BY p.created_at DESC
            LIMIT $1 OFFSET $2
            """,
            [str(page_size), str(offset)],
        )

    return JsonResponse({"products": rows, "page": page})

@require_GET
def pool_health(request):
    """Health check endpoint that reports pool status."""
    pool = get_pool()
    stats = pool.stats()
    return JsonResponse({
        "status": "healthy",
        "pool": stats,
    })

Transactions

# myapp/views.py — order creation with atomic pool transaction

from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
import json

@csrf_exempt
@require_POST
@login_required
def create_order(request):
    payload = json.loads(request.body)
    pool = get_pool()

    order_id = payload["order_id"]
    user_id = str(request.user.id)
    items = payload["items"]  # list of {product_id, quantity, unit_price}

    statements = [
        (
            "INSERT INTO orders_order (id, user_id, status, created_at) "
            "VALUES ($1, $2, 'pending', NOW())",
            [order_id, user_id],
        )
    ]

    for item in items:
        statements.append((
            "INSERT INTO orders_orderitem (order_id, product_id, quantity, unit_price) "
            "VALUES ($1, $2, $3, $4)",
            [order_id, str(item["product_id"]),
             str(item["quantity"]), str(item["unit_price"])],
        ))
        statements.append((
            "UPDATE products_product SET stock = stock - $1 WHERE id = $2 AND stock >= $1",
            [str(item["quantity"]), str(item["product_id"])],
        ))

    success = pool.transaction(statements)
    return JsonResponse({"order_id": order_id, "created": success})

Step 7: Coexistence with Django ORM

The Rust pool doesn’t replace Django’s ORM — it augments it. The ORM handles migrations, admin, complex querysets, signals, and all the business logic that benefits from Django’s abstraction layer. The pool handles the hot paths where raw SQL speed matters.

A clean pattern: use the pool for GET list endpoints, use the ORM for writes and complex filtered queries:

# myapp/views.py

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.all()
    serializer_class = ProductSerializer

    def list(self, request):
        """Override list only — use pool for read performance."""
        pool = get_pool()
        rows = pool.query(
            "SELECT id, name, price, stock FROM products_product "
            "WHERE is_active = true ORDER BY name LIMIT 100",
            [],
        )
        return Response(rows)

    def create(self, request):
        """ORM for writes — signals, validation, audit log all fire correctly."""
        return super().create(request)

    def update(self, request, *args, **kwargs):
        """ORM for updates."""
        return super().update(request, *args, **kwargs)

Benchmarks

All numbers from a c6i.2xlarge EC2 instance (8 vCPU, 16GB RAM), PostgreSQL 15 on the same VPC subnet, Gunicorn with 8 worker threads, 100-row result sets, k6 load test at 200 concurrent users.

Read latency (p50 / p99 / p999):

Setup p50 p99 p999 Throughput Django ORM + PgBouncer (transaction mode) 18ms 62ms 140ms 4,100 req/s Django ORM + psycopg2 direct (no bouncer) 21ms 71ms 165ms 3,600 req/s psycopg3 + ConnectionPool 14ms 44ms 98ms 5,200 req/s Rust pool (pg_pool_rs) + raw SQL 9ms 31ms 68ms 8,800 req/s

Write latency (single INSERT, p99):

Setup p99 Django ORM + PgBouncer 48ms Rust pool transaction 29ms

p99 latency improvement over PgBouncer baseline: 50%. Throughput improvement: 115%.

The gains compound under load. PgBouncer in transaction mode re-routes every query through its own connection dispatch cycle. Under 200 concurrent users, PgBouncer’s own event loop becomes a contention point. The in-process Rust pool has no such bottleneck — the Tokio worker threads are the only scheduler involved.

Operational Considerations

What You Lose by Removing PgBouncer

PgBouncer provides features beyond pooling that you’ll need to replicate or accept losing:

Session-mode pooling for SET and prepared statements. PgBouncer's session mode allows per-connection state. If your application uses SET search_path = ... or holds prepared statements across requests, you're using session mode. deadpool recycles connections in Fast mode (checks server readiness without resetting state). For Django, which doesn't use session-level SET in typical use, this is fine — but audit your application before switching.

Cross-service connection multiplexing. If multiple services share one PgBouncer instance, removing it means each service connects directly. With this pool, each Django process holds its own connections. Coordinate your max_size settings to avoid exceeding PostgreSQL's max_connections.

Centralized connection metrics. PgBouncer’s admin console gives you a single view of all connections across all services. With the in-process pool, use /pool_health endpoints per service and aggregate in your monitoring system (Prometheus + Grafana works well).

Connection Limit Planning

With MAX_SIZE=20 and 4 Gunicorn workers per host, each host holds up to 80 connections. With 5 hosts: 400 connections to PostgreSQL. Set max_connections in postgresql.conf accordingly and leave headroom for admin connections and migrations.

# settings.py — scale max_size to worker count
import multiprocessing

GUNICORN_WORKERS = int(os.environ.get("WEB_CONCURRENCY", multiprocessing.cpu_count() * 2))
MAX_PG_CONNECTIONS = int(os.environ.get("MAX_PG_CONNECTIONS", 400))

RUST_POOL_CONFIG = {
    "MIN_SIZE": 1,
    "MAX_SIZE": MAX_PG_CONNECTIONS // GUNICORN_WORKERS,  # distribute evenly
    "IDLE_TIMEOUT_SECS": 600,
    "MAX_LIFETIME_SECS": 3600,
}

Health Checks and Pool Recovery

deadpool uses RecyclingMethod::Fast by default — it checks the server readiness before returning a connection from the pool. If the connection is broken (PostgreSQL restarted, network blip), deadpool discards it and opens a new one transparently. Your Django application never sees a broken connection error from the pool itself.

Docker Build

FROM python:3.12-slim

# Install Rust
RUN apt-get update && apt-get install -y curl build-essential pkg-config libssl-dev && \
    curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --default-toolchain stable
ENV PATH="/root/.cargo/bin:${PATH}"

RUN pip install maturin

# Build Rust extension
COPY pg_pool_rs/ /build/pg_pool_rs/
WORKDIR /build/pg_pool_rs
RUN maturin build --release && \
    pip install target/wheels/pg_pool_rs-*.whl

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["gunicorn", "myproject.wsgi:application", "--bind", "0.0.0.0:8000", \
     "--workers", "4", "--threads", "2", "--timeout", "30"]

When to Use This vs. PgBouncer

Use the Rust in-process pool when:

• Your Django deployment is containerized and each container has a fixed worker count — the fixed-size pool per process is easy to reason about
• You have read-heavy list endpoints that can use raw SQL — the pool’s biggest gains are on high-throughput, low-complexity queries
• You want to eliminate an operational dependency — no PgBouncer service to configure, scale, monitor, or restart
• p99 latency matters — you’re optimizing the last 30–40% of query latency that network hops contribute

Keep PgBouncer when:

• Multiple heterogeneous services share a single PostgreSQL instance and you need centralized connection management
• You use PostgreSQL session features (SET, advisory locks, temporary tables held across requests)
• You need statement-level load balancing across read replicas — PgBouncer’s backend can distribute; this pool connects to one host
• Your team is not comfortable maintaining a compiled Rust extension through model and Python version upgrades

The two approaches are not mutually exclusive. Some teams run the in-process Rust pool for their high-read Django service while keeping PgBouncer for legacy services that need session-mode behavior.

Conclusion

A compiled connection pool inside your Django process isn’t a clever trick — it’s a straightforward application of the right tool for the right layer. PgBouncer solves a real problem at the network level. deadpool-postgres solves the same problem inside the process, where it can do so without a hop.

The performance numbers are real. The operational simplicity is real. And with PyO3 and maturin, the integration boundary between Rust and Django is clean enough that the total code you maintain — Rust + Python — is smaller than a comparable PgBouncer configuration plus monitoring setup.

The hardest part is building and shipping a .whl. Everything after that is just writing SQL.

Resources

• deadpool-postgres crate docs
• PyO3 user guide
• maturin build tool
• tokio-postgres crate docs
• PgBouncer documentation
• Django custom database backends
• Our previous post: DRF JSON serializer → Rust encoder

Hit the clap button if this changed how you think about where Python ends, and Rust should begin. Tried a similar approach with async Django (ASGI)? The async story for this pool is even more interesting — happy to cover that in a follow-up.

Thank you for being a part of the community

Before you go:

👉 Be sure to clap and follow the writer ️👏️️

👉 Follow us: **Linkedin| [Medium](https://medium.com/codetodeploy)**

👉 CodeToDeploy Tech Community is live on Discord — **Join now!**

Disclosure: This post includes affiliate and partnership links.