PostgreSQL Connection Pooling Best Practices¶
Objective: Master production-grade PostgreSQL connection pooling for high-concurrency workloads. When you need to handle thousands of concurrent requests, when you're running FastAPI with geospatial queries, when you're managing Prefect agents and Dask workers—PostgreSQL connection pooling becomes your lifeline.
Postgres will not forgive sloppy connection handling. Without pooling, you'll hit limits, crash services, and wonder why everything stalls.
0) Prerequisites (Read Once, Live by Them)¶
The Five Commandments¶
- Never connect directly to Postgres in production
- Use PgBouncer for external pooling
- Implement application-level pools
- Monitor connection usage
-
Set appropriate timeouts
-
Understand Postgres connection limits
- Each connection = backend process = memory overhead
- Default max_connections = 100 (too low for production)
- Each connection uses ~2-8MB RAM
-
Too many connections = OOM or TooManyConnectionsError
-
Use transaction pooling with PgBouncer
- Most efficient for web applications
- Reuses connections across transactions
- Keeps Postgres connection count low
-
Handles prepared statement caveats
-
Implement application-level pools
- One pool per service, not per request
- Use asyncpg.Pool or SQLAlchemy async pools
- Set appropriate min/max sizes
-
Monitor pool saturation
-
Monitor and secure everything
- Track connection metrics
- Use TLS for all connections
- Implement proper authentication
- Log connection events
Why These Principles: Production PostgreSQL requires understanding connection limits, pooling strategies, and monitoring patterns. Understanding these patterns prevents connection exhaustion and enables reliable database operations.
1) Why Connection Pooling Matters¶
PostgreSQL Connection Reality¶
-- Check current connections
SELECT count(*) FROM pg_stat_activity;
-- Check connection limits
SHOW max_connections;
-- Check memory usage per connection
SELECT
setting::int * 8192 / 1024 / 1024 as memory_mb
FROM pg_settings
WHERE name = 'max_connections';
Why Connection Limits Matter: PostgreSQL has a hard limit on concurrent connections. Each connection consumes memory and CPU resources. Understanding these limits prevents connection exhaustion and enables reliable database operations.
Connection Overhead¶
# Each connection costs memory
import psycopg2
import psutil
import os
# Check memory before connection
process = psutil.Process(os.getpid())
mem_before = process.memory_info().rss / 1024 / 1024
# Create connection
conn = psycopg2.connect("postgresql://user:pass@localhost/db")
mem_after = process.memory_info().rss / 1024 / 1024
print(f"Memory per connection: {mem_after - mem_before:.2f} MB")
Why Connection Overhead Matters: Each PostgreSQL connection consumes significant memory and CPU resources. Understanding these costs prevents resource exhaustion and enables efficient connection management.
2) PgBouncer Setup (Transaction Pooling)¶
Basic PgBouncer Configuration¶
# pgbouncer.ini
[databases]
mydb = host=postgres port=5432 dbname=mydb user=postgres password=secret
analytics = host=postgres port=5432 dbname=analytics user=analytics password=secret
[pgbouncer]
listen_port = 6432
listen_addr = 0.0.0.0
auth_type = md5
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 50
reserve_pool_size = 10
reserve_pool_timeout = 5
max_db_connections = 100
max_user_connections = 50
server_round_robin = 1
ignore_startup_parameters = extra_float_digits
Why PgBouncer Configuration Matters: Proper PgBouncer setup enables efficient connection pooling and prevents connection exhaustion. Understanding these patterns prevents database bottlenecks and enables reliable connection management.
Docker Compose Integration¶
# docker-compose.yml
version: "3.9"
services:
postgres:
image: postgres:17
environment:
POSTGRES_DB: mydb
POSTGRES_USER: postgres
POSTGRES_PASSWORD: secret
ports: ["5432:5432"]
volumes:
- postgres-data:/var/lib/postgresql/data
pgbouncer:
image: edoburu/pgbouncer:1.21.0
environment:
DATABASES_HOST: postgres
DATABASES_PORT: 5432
DATABASES_USER: postgres
DATABASES_PASSWORD: secret
DATABASES_DBNAME: mydb
POOL_MODE: transaction
MAX_CLIENT_CONN: 1000
DEFAULT_POOL_SIZE: 50
RESERVE_POOL_SIZE: 10
RESERVE_POOL_TIMEOUT: 5
ports: ["6432:6432"]
depends_on: [postgres]
app:
build: .
environment:
DATABASE_URL: postgresql://user:pass@pgbouncer:6432/mydb
depends_on: [pgbouncer]
volumes:
postgres-data:
Why Docker Integration Matters: Containerized PgBouncer enables consistent deployment and scaling. Understanding these patterns prevents deployment chaos and enables reliable connection pooling.
Advanced PgBouncer Configuration¶
# Advanced pgbouncer.ini
[databases]
# Primary database
mydb = host=postgres port=5432 dbname=mydb user=postgres password=secret
# Read replica
mydb_ro = host=postgres-ro port=5432 dbname=mydb user=readonly password=secret
[pgbouncer]
listen_port = 6432
listen_addr = 0.0.0.0
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
pool_mode = transaction
max_client_conn = 2000
default_pool_size = 100
reserve_pool_size = 20
reserve_pool_timeout = 10
max_db_connections = 200
max_user_connections = 100
server_round_robin = 1
ignore_startup_parameters = extra_float_digits,application_name
log_connections = 1
log_disconnections = 1
log_pooler_errors = 1
stats_period = 60
Why Advanced Configuration Matters: Production PgBouncer requires careful tuning for performance and reliability. Understanding these patterns prevents connection bottlenecks and enables efficient database access.
3) Application-Level Pools (Python asyncpg)¶
AsyncPG Pool Configuration¶
# asyncpg_pool.py
import asyncpg
import asyncio
from typing import Optional
class DatabasePool:
def __init__(self, dsn: str, min_size: int = 5, max_size: int = 20):
self.dsn = dsn
self.min_size = min_size
self.max_size = max_size
self.pool: Optional[asyncpg.Pool] = None
async def create_pool(self):
"""Create connection pool"""
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=self.min_size,
max_size=self.max_size,
command_timeout=30,
server_settings={
'application_name': 'myapp',
'timezone': 'UTC'
}
)
async def close_pool(self):
"""Close connection pool"""
if self.pool:
await self.pool.close()
async def execute_query(self, query: str, *args):
"""Execute query with connection from pool"""
async with self.pool.acquire() as conn:
return await conn.fetch(query, *args)
async def execute_transaction(self, queries: list):
"""Execute multiple queries in transaction"""
async with self.pool.acquire() as conn:
async with conn.transaction():
results = []
for query, args in queries:
result = await conn.fetch(query, *args)
results.append(result)
return results
# Usage example
async def main():
pool = DatabasePool(
dsn="postgresql://user:pass@pgbouncer:6432/mydb",
min_size=5,
max_size=20
)
await pool.create_pool()
# Execute query
rows = await pool.execute_query("SELECT * FROM users WHERE id = $1", 1)
print(rows)
await pool.close_pool()
if __name__ == "__main__":
asyncio.run(main())
Why AsyncPG Pools Matter: Application-level pools reduce connection overhead and enable efficient database access. Understanding these patterns prevents connection leaks and enables reliable database operations.
FastAPI Integration¶
# fastapi_pool.py
from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
import asyncpg
from typing import AsyncGenerator
# Global pool
pool: asyncpg.Pool = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan manager"""
global pool
# Create pool on startup
pool = await asyncpg.create_pool(
"postgresql://user:pass@pgbouncer:6432/mydb",
min_size=5,
max_size=20
)
yield
# Close pool on shutdown
await pool.close()
app = FastAPI(lifespan=lifespan)
async def get_db_connection():
"""Dependency to get database connection"""
async with pool.acquire() as conn:
yield conn
@app.get("/users/{user_id}")
async def get_user(user_id: int, conn = Depends(get_db_connection)):
"""Get user by ID"""
row = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
return dict(row) if row else None
@app.post("/users")
async def create_user(name: str, email: str, conn = Depends(get_db_connection)):
"""Create new user"""
row = await conn.fetchrow(
"INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *",
name, email
)
return dict(row)
Why FastAPI Integration Matters: Proper FastAPI pooling enables efficient request handling and database access. Understanding these patterns prevents connection leaks and enables reliable web applications.
4) SQLAlchemy 2.x Async Pools¶
SQLAlchemy Async Engine Configuration¶
# sqlalchemy_pool.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text
import asyncio
# Create async engine with pool
engine = create_async_engine(
"postgresql+asyncpg://user:pass@pgbouncer:6432/mydb",
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=3600,
echo=False
)
# Create async session factory
AsyncSessionLocal = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
async def get_db_session():
"""Get database session"""
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
# Usage example
async def main():
async with AsyncSessionLocal() as session:
# Execute raw SQL
result = await session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1})
rows = result.fetchall()
print(rows)
# Use ORM
from sqlalchemy import select
from models import User
stmt = select(User).where(User.id == 1)
result = await session.execute(stmt)
user = result.scalar_one_or_none()
print(user)
if __name__ == "__main__":
asyncio.run(main())
Why SQLAlchemy Pools Matter: SQLAlchemy async pools enable efficient ORM operations and connection management. Understanding these patterns prevents connection leaks and enables reliable database operations.
SQLAlchemy with FastAPI¶
# fastapi_sqlalchemy.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models import User
from database import get_db_session
app = FastAPI()
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db_session)):
"""Get user by ID using SQLAlchemy"""
stmt = select(User).where(User.id == user_id)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
return user
@app.post("/users")
async def create_user(name: str, email: str, db: AsyncSession = Depends(get_db_session)):
"""Create new user using SQLAlchemy"""
user = User(name=name, email=email)
db.add(user)
await db.commit()
await db.refresh(user)
return user
Why SQLAlchemy FastAPI Integration Matters: Proper SQLAlchemy integration enables efficient ORM operations and connection management. Understanding these patterns prevents connection leaks and enables reliable web applications.
5) When to Use Which Pooling Strategy¶
Pooling Strategy Decision Matrix¶
# Pooling strategy decision matrix
POOLING_STRATEGIES = {
"pgbouncer_only": {
"use_case": "Many microservices hitting Postgres",
"pros": ["Centralized pooling", "Reduces Postgres connections", "Easy to monitor"],
"cons": ["Single point of failure", "Network latency", "Prepared statement issues"],
"config": "transaction pooling, 50-100 pool size"
},
"app_level_only": {
"use_case": "Single application with controlled load",
"pros": ["No network overhead", "Full control", "Better error handling"],
"cons": ["Per-app connection overhead", "Harder to monitor", "No cross-service pooling"],
"config": "asyncpg.Pool, 10-50 pool size"
},
"hybrid": {
"use_case": "Production microservices with high concurrency",
"pros": ["Best of both worlds", "Scalable", "Fault tolerant"],
"cons": ["Complex setup", "More monitoring", "Higher resource usage"],
"config": "PgBouncer + app pools, 20-50 app pool, 100-200 PgBouncer pool"
}
}
Why Pooling Strategy Matters: Choosing the right pooling strategy depends on your architecture and requirements. Understanding these patterns prevents connection bottlenecks and enables efficient database access.
Recommended Configurations¶
# Production configurations by use case
configurations:
high_concurrency:
pgbouncer:
pool_mode: "transaction"
max_client_conn: 2000
default_pool_size: 100
app_pool:
min_size: 10
max_size: 50
microservices:
pgbouncer:
pool_mode: "transaction"
max_client_conn: 1000
default_pool_size: 50
app_pool:
min_size: 5
max_size: 20
single_app:
pgbouncer: "disabled"
app_pool:
min_size: 10
max_size: 100
Why Configuration Matters: Proper configuration prevents connection exhaustion and enables efficient database access. Understanding these patterns prevents database bottlenecks and enables reliable connection management.
6) Scaling Patterns¶
One Pool Per Service¶
# service_pool.py
class ServicePool:
def __init__(self, service_name: str, dsn: str):
self.service_name = service_name
self.dsn = dsn
self.pool = None
async def initialize(self):
"""Initialize pool for this service"""
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=5,
max_size=20,
command_timeout=30
)
async def cleanup(self):
"""Cleanup pool for this service"""
if self.pool:
await self.pool.close()
# Usage in different services
user_service_pool = ServicePool("user_service", "postgresql://...")
order_service_pool = ServicePool("order_service", "postgresql://...")
Why One Pool Per Service Matters: Service-specific pools prevent connection conflicts and enable independent scaling. Understanding these patterns prevents connection chaos and enables reliable microservices.
Idle Timeout Configuration¶
# idle_timeout_config.py
import asyncio
from datetime import datetime, timedelta
class PoolWithIdleTimeout:
def __init__(self, dsn: str, idle_timeout: int = 300):
self.dsn = dsn
self.idle_timeout = idle_timeout
self.pool = None
self.last_used = datetime.now()
async def create_pool(self):
"""Create pool with idle timeout"""
self.pool = await asyncpg.create_pool(
self.dsn,
min_size=5,
max_size=20,
command_timeout=30
)
# Start idle timeout monitor
asyncio.create_task(self._monitor_idle_timeout())
async def _monitor_idle_timeout(self):
"""Monitor and cleanup idle connections"""
while True:
await asyncio.sleep(60) # Check every minute
if datetime.now() - self.last_used > timedelta(seconds=self.idle_timeout):
if self.pool:
await self.pool.close()
self.pool = None
print("Pool closed due to idle timeout")
async def acquire_connection(self):
"""Acquire connection and update last used time"""
if not self.pool:
await self.create_pool()
self.last_used = datetime.now()
return self.pool.acquire()
Why Idle Timeout Matters: Idle timeouts prevent zombie connections and resource leaks. Understanding these patterns prevents connection exhaustion and enables efficient resource management.
7) Monitoring and Observability¶
PgBouncer Metrics¶
# pgbouncer_monitoring.py
import asyncpg
import asyncio
from typing import Dict, Any
class PgBouncerMonitor:
def __init__(self, dsn: str):
self.dsn = dsn
self.conn = None
async def connect(self):
"""Connect to PgBouncer admin interface"""
self.conn = await asyncpg.connect(self.dsn)
async def get_stats(self) -> Dict[str, Any]:
"""Get PgBouncer statistics"""
if not self.conn:
await self.connect()
# Get pool statistics
stats = await self.conn.fetch("SHOW POOLS")
clients = await self.conn.fetch("SHOW CLIENTS")
servers = await self.conn.fetch("SHOW SERVERS")
return {
"pools": [dict(row) for row in stats],
"clients": [dict(row) for row in clients],
"servers": [dict(row) for row in servers]
}
async def get_pool_usage(self) -> Dict[str, Any]:
"""Get pool usage statistics"""
stats = await self.get_stats()
pool_usage = {}
for pool in stats["pools"]:
database = pool["database"]
pool_usage[database] = {
"clients_active": pool["cl_active"],
"clients_waiting": pool["cl_waiting"],
"servers_active": pool["sv_active"],
"servers_idle": pool["sv_idle"],
"servers_used": pool["sv_used"],
"servers_tested": pool["sv_tested"],
"servers_login": pool["sv_login"],
"max_wait": pool["maxwait"],
"pool_mode": pool["pool_mode"]
}
return pool_usage
# Usage
async def main():
monitor = PgBouncerMonitor("postgresql://user:pass@pgbouncer:6432/pgbouncer")
stats = await monitor.get_stats()
usage = await monitor.get_pool_usage()
print("Pool Usage:", usage)
print("Statistics:", stats)
if __name__ == "__main__":
asyncio.run(main())
Why Monitoring Matters: Pool monitoring enables proactive connection management and prevents bottlenecks. Understanding these patterns prevents connection exhaustion and enables reliable database operations.
Prometheus Integration¶
# prometheus_monitoring.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import asyncio
import time
# Prometheus metrics
connection_requests = Counter('postgres_connection_requests_total', 'Total connection requests')
connection_duration = Histogram('postgres_connection_duration_seconds', 'Connection duration')
active_connections = Gauge('postgres_active_connections', 'Active connections')
pool_size = Gauge('postgres_pool_size', 'Pool size')
class PrometheusPoolMonitor:
def __init__(self, pool):
self.pool = pool
self.start_time = time.time()
async def record_connection_usage(self):
"""Record connection usage metrics"""
while True:
# Record active connections
active_connections.set(self.pool.get_size())
pool_size.set(self.pool.get_max_size())
await asyncio.sleep(10) # Update every 10 seconds
async def execute_with_metrics(self, query: str, *args):
"""Execute query with metrics"""
connection_requests.inc()
start_time = time.time()
async with self.pool.acquire() as conn:
result = await conn.fetch(query, *args)
connection_duration.observe(time.time() - start_time)
return result
# Start Prometheus server
start_http_server(8000)
Why Prometheus Integration Matters: Metrics collection enables monitoring and alerting on connection pool health. Understanding these patterns prevents connection issues and enables reliable database operations.
8) Security and Best Practices¶
TLS Configuration¶
# pgbouncer.ini with TLS
[databases]
mydb = host=postgres port=5432 dbname=mydb user=postgres password=secret
[pgbouncer]
listen_port = 6432
listen_addr = 0.0.0.0
auth_type = md5
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 50
# TLS configuration
client_tls_sslmode = require
client_tls_cert_file = /etc/ssl/certs/pgbouncer.crt
client_tls_key_file = /etc/ssl/private/pgbouncer.key
client_tls_ca_file = /etc/ssl/certs/ca.crt
server_tls_sslmode = require
server_tls_cert_file = /etc/ssl/certs/pgbouncer.crt
server_tls_key_file = /etc/ssl/private/pgbouncer.key
server_tls_ca_file = /etc/ssl/certs/ca.crt
Why TLS Configuration Matters: Secure connections prevent data interception and enable compliance. Understanding these patterns prevents security vulnerabilities and enables reliable database access.
Authentication and Authorization¶
# userlist.txt
"postgres" "md5a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6"
"app_user" "md5x1y2z3a4b5c6d7e8f9g0h1i2j3k4l5"
"readonly_user" "md5m1n2o3p4q5r6s7t8u9v0w1x2y3z4"
Why Authentication Matters: Proper authentication prevents unauthorized access and enables audit trails. Understanding these patterns prevents security breaches and enables reliable database access.
Prepared Statement Caveats¶
# prepared_statements_caveats.py
import asyncpg
async def problematic_prepared_statements():
"""Example of problematic prepared statements with PgBouncer"""
pool = await asyncpg.create_pool(
"postgresql://user:pass@pgbouncer:6432/mydb",
min_size=5,
max_size=20
)
async with pool.acquire() as conn:
# This can cause issues with transaction pooling
await conn.execute("PREPARE stmt AS SELECT * FROM users WHERE id = $1")
result = await conn.fetch("EXECUTE stmt(1)")
# Better approach: use parameterized queries
result = await conn.fetch("SELECT * FROM users WHERE id = $1", 1)
async def safe_prepared_statements():
"""Safe approach to prepared statements"""
pool = await asyncpg.create_pool(
"postgresql://user:pass@pgbouncer:6432/mydb",
min_size=5,
max_size=20
)
async with pool.acquire() as conn:
# Use parameterized queries instead of prepared statements
result = await conn.fetch("SELECT * FROM users WHERE id = $1", 1)
return result
Why Prepared Statement Caveats Matter: Transaction pooling can cause issues with prepared statements. Understanding these patterns prevents connection errors and enables reliable database operations.
9) Architecture Diagram¶
Connection Pooling Flow¶
flowchart LR
A[FastAPI App] -->|Connection Request| B[App Pool]
B -->|Pooled Connection| C[PgBouncer]
C -->|Transaction Pool| D[PostgreSQL]
E[Prefect Agent] -->|Connection Request| F[App Pool]
F -->|Pooled Connection| C
G[Dask Worker] -->|Connection Request| H[App Pool]
H -->|Pooled Connection| C
subgraph "Application Layer"
B
F
H
end
subgraph "Pooling Layer"
C
end
subgraph "Database Layer"
D
end
subgraph "Monitoring"
I[Prometheus Metrics]
J[Connection Stats]
K[Pool Usage]
end
B --> I
F --> I
H --> I
C --> J
C --> K Why Architecture Diagrams Matter: Visual representation of connection flow enables understanding of pooling strategies. Understanding these patterns prevents connection confusion and enables reliable database architecture.
10) TL;DR Runbook¶
Essential Commands¶
# Start PgBouncer
pgbouncer -d pgbouncer.ini
# Check PgBouncer status
psql -h localhost -p 6432 -U pgbouncer pgbouncer -c "SHOW POOLS"
# Monitor connections
psql -h localhost -p 6432 -U pgbouncer pgbouncer -c "SHOW CLIENTS"
# Check PostgreSQL connections
psql -h localhost -p 5432 -U postgres -c "SELECT count(*) FROM pg_stat_activity"
Essential Patterns¶
# Essential PostgreSQL pooling patterns
postgres_pooling_patterns:
"never_direct": "Never connect directly from app to Postgres in production",
"always_pgbouncer": "Always run PgBouncer (transaction pooling)",
"app_level_pools": "Use app-level pools (asyncpg, SQLAlchemy) but keep them small",
"monitor_saturation": "Monitor pool saturation and timeouts",
"prepared_statements": "Watch out for prepared statements + PgBouncer mode",
"secure_connections": "Secure connections with TLS + proper creds",
"one_pool_per_service": "One pool per service, not per request",
"idle_timeout": "Use idle timeout cleanup",
"monitor_metrics": "Monitor pool saturation and connection usage"
Quick Reference¶
# Essential PostgreSQL pooling operations
# 1. PgBouncer configuration
[pgbouncer]
pool_mode = transaction
max_client_conn = 1000
default_pool_size = 50
# 2. AsyncPG pool
pool = await asyncpg.create_pool(
"postgresql://user:pass@pgbouncer:6432/mydb",
min_size=5, max_size=20
)
# 3. SQLAlchemy async pool
engine = create_async_engine(
"postgresql+asyncpg://user:pass@pgbouncer:6432/mydb",
pool_size=20, max_overflow=10
)
# 4. Connection usage
async with pool.acquire() as conn:
result = await conn.fetch("SELECT * FROM users")
# 5. Monitoring
stats = await conn.fetch("SHOW POOLS")
Why This Runbook: These patterns cover 90% of PostgreSQL pooling needs. Master these before exploring advanced connection management scenarios.
11) The Machine's Summary¶
PostgreSQL connection pooling requires understanding connection limits, pooling strategies, and monitoring patterns. When used correctly, connection pooling enables reliable database operations, prevents connection exhaustion, and supports high-concurrency workloads. The key is understanding Postgres connection overhead, implementing proper pooling strategies, and monitoring connection health.
The Dark Truth: Without proper connection pooling, your database will choke on connection limits and your applications will fail silently. Connection pooling is your lifeline. Use it wisely.
The Machine's Mantra: "In the pooling we trust, in the monitoring we find reliability, and in the security we find the path to production-grade database operations."
Why This Matters: PostgreSQL connection pooling enables efficient database access that can handle high-concurrency workloads, prevent connection exhaustion, and provide reliable database operations while ensuring security and monitoring.
See Also¶
- PostgreSQL Pooling Tutorial - Step-by-step implementation guide with PgBouncer and FastAPI
- Database Optimization - PostgreSQL and PostGIS performance tuning
- PostgreSQL Monitoring & Observability - Database monitoring and observability patterns
This guide provides the complete machinery for PostgreSQL connection pooling. The patterns scale from simple applications to complex microservices, from basic pooling to advanced connection management.