hasql

hasql is a library for acquiring actual connections to masters and replicas in high available PostgreSQL clusters.

Features

• completely asynchronous api
• automatic detection of the host role in the cluster
• health-checks for each host and automatic traffic outage for unavailable hosts
• autodetection of hosts role changes, in case replica host will be promoted to master
• different policies for load balancing
• support for asyncpg, psycopg3, aiopg, sqlalchemy and asyncpgsa

Usage

Some useful examples

Creating connection pool

When acquiring a connection, the connection object of the used driver is returned (aiopg.connection.Connection for aiopg and asyncpg.pool.PoolConnectionProxy for asyncpg and asyncpgsa)

Database URL specirication rules

• Multiple hosts should be passed comma separated
  • multihost example:
    • postgresql://db1,db2,db3/
  • split result:
    • postgresql://db1:5432/
    • postgresql://db2:5432/
    • postgresql://db3:5432/
• The non-default port for each host might be passed after hostnames. e.g.
  • multihost example:
    • postgresql://db1:1234,db2:5678,db3/
  • split result:
    • postgresql://db1:1234/
    • postgresql://db2:5678/
    • postgresql://db3:5432/
• The special case for non-default port for all hosts
  • multihost example:
    • postgresql://db1,db2,db3:6432/
  • split result:
    • postgresql://db1:6432/
    • postgresql://db2:6432/
    • postgresql://db3:6432/

For aiopg or aiopg.sa

aiopg must be installed as a requirement.

Code example using aiopg:

from hasql.driver.aiopg import PoolManager

hosts = ",".join([
    "master-host:5432",
    "replica-host-1:5432",
    "replica-host-2:5432",
])

multihost_dsn = f"postgresql://user:password@{hosts}/dbname"

async def create_pool(dsn) -> PoolManager:
    pool = PoolManager(multihost_dsn)

    # Waiting for 1 master and 1 replica will be available
    await pool.ready(masters_count=1, replicas_count=1)
    return pool

Code example using aiopg.sa:

from hasql.driver.aiopg_sa import PoolManager

hosts = ",".join([
    "master-host:5432",
    "replica-host-1:5432",
    "replica-host-2:5432",
])

multihost_dsn = f"postgresql://user:password@{hosts}/dbname"

async def create_pool(dsn) -> PoolManager:
    pool = PoolManager(multihost_dsn)

    # Waiting for 1 master and 1 replica will be available
    await pool.ready(masters_count=1, replicas_count=1)
    return pool

For asyncpg

asyncpg must be installed as a requirement

from hasql.driver.asyncpg import PoolManager

hosts = ",".join([
    "master-host:5432",
    "replica-host-1:5432",
    "replica-host-2:5432",
])

multihost_dsn = f"postgresql://user:password@{hosts}/dbname"

async def create_pool(dsn) -> PoolManager:
    pool = PoolManager(multihost_dsn)

    # Waiting for 1 master and 1 replica will be available
    await pool.ready(masters_count=1, replicas_count=1)
    return pool

For sqlalchemy

sqlalchemy[asyncio] & asyncpg must be installed as requirements

from hasql.driver.asyncsqlalchemy import PoolManager

hosts = ",".join([
    "master-host:5432",
    "replica-host-1:5432",
    "replica-host-2:5432",
])

multihost_dsn = f"postgresql://user:password@{hosts}/dbname"


async def create_pool(dsn) -> PoolManager:
    pool = PoolManager(
        multihost_dsn,

        # Use master for acquire_replica, if no replicas available
        fallback_master=True,

        # You can pass pool-specific options
        pool_factory_kwargs=dict(
            pool_size=10,
            max_overflow=5
        )
    )

    # Waiting for 1 master and 1 replica will be available
    await pool.ready(masters_count=1, replicas_count=1)
    return pool

For asyncpgsa

asyncpgsa must be installed as a requirement

from hasql.driver.asyncpgsa import PoolManager

hosts = ",".join([
    "master-host:5432",
    "replica-host-1:5432",
    "replica-host-2:5432",
])

multihost_dsn = f"postgresql://user:password@{hosts}/dbname"

async def create_pool(dsn) -> PoolManager:
    pool = PoolManager(multihost_dsn)

    # Waiting for 1 master and 1 replica will be available
    await pool.ready(masters_count=1, replicas_count=1)
    return pool

For psycopg3

psycopg3 must be installed as a requirement (package name is psycopg) Configure queue limits explicitly with pool_factory_kwargs={"max_waiting": ...} if you want psycopg_pool.TooManyRequests on pool saturation. Otherwise the driver default queue behavior is used.

from hasql.driver.psycopg3 import PoolManager


hosts = ",".join([
    "master-host:5432",
    "replica-host-1:5432",
    "replica-host-2:5432",
])
multihost_dsn = f"postgresql://user:password@{hosts}/dbname"

async def create_pool(dsn) -> PoolManager:
    pool = PoolManager(multihost_dsn)

    # Waiting for 1 master and 1 replica will be available
    await pool.ready(masters_count=1, replicas_count=1)
    return pool

Acquiring connections

Connections should be acquired with async context manager:

Acquiring master connection

async def do_something():
    pool = await create_pool(multihost_dsn)
    async with pool.acquire(read_only=False) as connection:
        ...

or

async def do_something():
    pool = await create_pool(multihost_dsn)
    async with pool.acquire_master() as connection:
        ...

Acquiring replica connection

async def do_something():
    pool = await create_pool(multihost_dsn)
    async with pool.acquire(read_only=True) as connection:
        ...

or

async def do_something():
    pool = await create_pool(multihost_dsn)
    async with pool.acquire_replica() as connection:
        ...

How it works?

For each host from dsn string, a connection pool is created. From each pool one connection is reserved, which is used to check the availability of the host and its role. The minimum and maximum number of connections in the pool increases by 1 (to reserve a system connection).

For each pool a background task is created, in which the host availability and its role (master or replica) is checked once every refresh_delay second.

When switching hosts roles, hasql detects this with a slight delay.

For PostgreSQL, when switching the master, all connections to all hosts are broken (the details of implementing PostgreSQL).

If there are no available hosts, the methods acquire(), acquire_master(), and acquire_replica() wait until the host with the desired role startup.

Balancer Policies

When multiple pools match the requested role (e.g. several healthy replicas), hasql uses a balancer policy to choose which pool to acquire a connection from. The policy is set via the balancer_policy parameter of PoolManager.

GreedyBalancerPolicy (default)

Picks the pool with the most free connections. When several pools are tied, chooses randomly among them.

Best for workloads where you want to fill up idle pools first and avoid acquiring from pools that are already under pressure.

RoundRobinBalancerPolicy

Cycles through available pools in order, giving each pool an equal share of requests regardless of pool state or host performance.

Best for uniform workloads where all replicas have similar hardware and you want simple, predictable distribution.

RandomWeightedBalancerPolicy

Uses random.choices with probabilities proportional to the inverse of each candidate's last health-check latency (P ∝ 1 / latency). The finite, scale-equivalent weights are relative and uncapped: lower valid latency gives a higher statistical selection probability. The latency is the health-check round-trip only; it does not represent query latency or pool load. If any candidate timing is missing or invalid (None, <= 0, NaN, or infinite), all candidates receive uniform weights. This is statistical selection, not a deterministic or fair scheduler, and provides no no-starvation guarantee.

Policy Comparison

Property	Greedy	RoundRobin	RandomWeighted
Selection strategy	Most free connections	Sequential rotation	Inverse health-check latency (lower latency = higher probability)
Adapts to load	Yes (pool state)	No	No (health-check latency only)
Thundering herd risk	Higher	None	Not eliminated; statistical selection
Heterogeneous replicas	Poor	Poor	Favors lower-latency replicas; not a pool-load signal
Predictability	Low	High	Statistical, not deterministic or fair
Best for	Low-concurrency	Uniform clusters	Latency-diverse replicas needing random weighting

from hasql.balancer_policy import (
    GreedyBalancerPolicy,
    RandomWeightedBalancerPolicy,
    RoundRobinBalancerPolicy,
)
from hasql.driver.asyncpg import PoolManager

pool = PoolManager(
    dsn,
    balancer_policy=RandomWeightedBalancerPolicy,
)

Metrics

Every PoolManager exposes a metrics() method that returns a point-in-time snapshot of the entire cluster state.

m = pool_manager.metrics()

The returned Metrics object contains three layers:

m.pools — per-pool metrics

A sequence of PoolMetrics dataclasses, one per database host:

Field	Description
host	Host address of the pool
role	"master", "replica", or None (unknown)
healthy	True if the host has a known role
min	Minimum connections configured
max	Maximum connections configured
idle	Connections currently idle in the pool
used	Connections currently checked out
response_time	Last health-check round-trip time (seconds)
in_flight	Connections acquired through the pool manager
extra	Driver-specific data (e.g. psycopg3's requests_waiting, SQLAlchemy's overflow)

m.gauges — cluster-wide gauges

A HasqlGauges dataclass with aggregate state:

Field	Description
master_count	Number of detected masters
replica_count	Number of detected replicas
available_count	Total pools with a known role
unavailable_count	Number of pools currently unavailable
stale_count	Number of pools currently classified as stale
active_connections	Connections currently held by application code
closing	True while the pool manager is shutting down
closed	True after shutdown is complete

m.hasql — internal counters

A HasqlMetrics dataclass with cumulative acquire/release counters and timing data, useful for tracking pool manager overhead.

Example: simple metrics endpoint

from dataclasses import asdict
import json

async def handle_metrics(request):
    m = pool_manager.metrics()
    return web.json_response(asdict(m))

Exporting metrics to OTLP

hasql ships with ready-to-use examples for exporting metrics to any OpenTelemetry-compatible collector (Prometheus, Grafana, Datadog, etc.) via OTLP gRPC.

Quick start

Install the OpenTelemetry dependencies:

pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc

Use the helper from example/otlp/common.py:

import asyncio

from hasql.driver.asyncpg import PoolManager
from example.otlp.common import (
    observe_hasql_metrics,
    setup_meter_provider,
)

async def main(dsn):
    provider = setup_meter_provider(export_interval_ms=10_000)
    try:
        pool = PoolManager(dsn, fallback_master=True)
        try:
            await pool.ready()
            async with observe_hasql_metrics(pool, sample_interval=1.0):
                while True:
                    async with pool.acquire_master() as conn:
                        await conn.fetchval("SELECT 1")
                    await asyncio.sleep(1)
        finally:
            await pool.close()
    finally:
        await asyncio.to_thread(provider.shutdown)

The helper samples metrics() once on the owning event loop before registration and then every sample_interval seconds (default: 1). OTel callbacks read only a detached immutable snapshot, never the live manager. Export runs independently (default: 10 seconds); both intervals must be positive and finite. Manual collection also reads the latest sample, not live state. Each callback retains one snapshot; a collection across instruments is not an atomic batch. A blocked event loop delays sampling. Sampling errors are logged and clear observations until the next successful sample. Acquire counters stay cumulative; missing lag and selected extra keys produce no observation.

Pass extra_keys=("overflow",) for SQLAlchemy extras, or selected psycopg3 keys, to the same helper. It stops and awaits its sampler before pool cleanup; the provider is owned by the caller and shut down off-loop even if cleanup fails. Final SDK collection may use the last detached sample after sampling stops. Export calls have a 10-second timeout; SDK shutdown defaults to 30 seconds.

Run the scripts as modules from the repository checkout (direct script paths can shadow driver packages):

python -m example.otlp.asyncpg --dsn postgresql://u:p@db1,db2/mydb

Exported OTel instruments

Gauge name	Labels	Source
db.pool.connections.min	host, role	PoolMetrics.min
db.pool.connections.max	host, role	PoolMetrics.max
db.pool.connections.idle	host, role	PoolMetrics.idle
db.pool.connections.used	host, role	PoolMetrics.used
db.pool.connections.in_flight	host, role	PoolMetrics.in_flight
db.pool.healthy	host, role	PoolMetrics.healthy
db.pool.health_check.duration	host, role	PoolMetrics.response_time
db.pool.masters	—	HasqlGauges.master_count
db.pool.replicas	—	HasqlGauges.replica_count
db.pool.active_connections	—	HasqlGauges.active_connections
db.pool.acquire.count	host	HasqlMetrics.acquire[host]
db.pool.acquire.duration	host	HasqlMetrics.acquire_time[host]
db.pool.stale.count	—	HasqlGauges.stale_count
db.pool.stale.status	host, role, staleness	PoolMetrics.staleness
db.pool.stale.lag.bytes	host, role, staleness	PoolMetrics.lag[\"bytes\"]
db.pool.stale.lag.time	host, role, staleness	PoolMetrics.lag[\"time\"] in seconds
db.pool.extra.<key>	host, role, optional staleness	PoolMetrics.extra[key]

Driver-specific extras

Some drivers expose additional pool internals via PoolMetrics.extra. Pass extra_keys to the observation context in the quick start above:

# psycopg3: queue depth, error counters, etc.
extra_keys = ("pool_size", "requests_waiting", "connections_errors")

# Or, for SQLAlchemy: overflow connections
extra_keys = ("overflow",)

async with observe_hasql_metrics(pool, extra_keys=extra_keys):
    ...  # application workload

Per-driver examples live in example/otlp/.

Dashboard recommendations

The exported metrics map well to Grafana / Datadog dashboard panels:

Cluster health overview

• db.pool.masters / db.pool.replicas — single-stat panels; alert when master drops to 0 or replicas drop below expected count
• db.pool.healthy by host — table or status map showing per-host health; any 0 value means the host lost its role

Connection pool utilization

• db.pool.connections.used / db.pool.connections.max by host — saturation ratio; alert when approaching 100%
• db.pool.connections.idle by host — if consistently 0, the pool is undersized
• db.pool.connections.in_flight by host — connections held by application code right now; spikes indicate slow queries or leaked connections

Acquisition and staleness

• db.pool.acquire.count and db.pool.acquire.duration — cumulative observable counters by host (duration is in seconds)
• db.pool.stale.count — point-in-time stale pool count
• db.pool.stale.status — per-pool status with string host, role, and staleness attributes
• db.pool.stale.lag.bytes and db.pool.stale.lag.time — latest byte and time lag; time lag is exported in seconds

Latency and performance

• db.pool.health_check.duration by host — time series; rising latency on a replica can predict upcoming failover
• Compare response_time across hosts to spot slow replicas before they affect user traffic

Pool manager overhead

• db.pool.active_connections — total connections held across all pools; correlate with application request rate to right-size pools

Driver-specific panels (psycopg3)

• db.pool.extra.requests_waiting — queue depth; sustained > 0 means the pool is saturated
• db.pool.extra.connections_errors — connection failures; alert on rate increase

Alerting rules

• db.pool.masters == 0 — critical: no master available
• db.pool.replicas == 0 — warning: no fresh replicas; reads use an available master if fallback is enabled, otherwise an available known stale replica. With no candidates, acquisition waits up to its timeout. Separately, master_as_replica_weight can include a master alongside fresh replicas
• db.pool.connections.used / db.pool.connections.max > 0.9 — warning: pool near exhaustion
• db.pool.health_check.duration > threshold — warning: host becoming slow, may lose role soon
• db.pool.extra.requests_waiting > 0 for sustained period — warning: pool undersized for current load

Replica staleness

Configure replica filtering with a StalenessPolicy. Byte lag compares a replica replay LSN with recently collected master state; time lag reads the replica replay timestamp directly.

from datetime import timedelta

from hasql.driver.asyncpg import PoolManager
from hasql.staleness import (
    BytesStalenessChecker,
    StalenessPolicy,
    TimeStalenessChecker,
)

by_bytes = StalenessPolicy(
    BytesStalenessChecker(
        max_lag_bytes=16 * 1024 * 1024,
        max_master_lsn_age=timedelta(seconds=2),
    ),
    grace_period=timedelta(seconds=5),
)
by_time = StalenessPolicy(
    TimeStalenessChecker(max_lag=timedelta(seconds=10)),
)
pool = PoolManager(dsn, staleness=by_bytes)

A stale result remains eligible during grace_period only when that pool was observed fresh recently. With no grace period it is removed immediately. Both checkers cache fresh master WAL LSN state; its default maximum age is two seconds. TimeStalenessChecker reports zero lag when a replica's replay LSN matches the cached master LSN, and otherwise evaluates replay-timestamp delay. Missing or expired master state fails open with empty lag, consistently with the byte checker. Query errors fail closed through the health monitor: the pool is removed from every availability set until a later successful health check. Because time lag is calculated from wall-clock timestamps, clock skew can affect reported values for a replica that is behind. Read acquisition prioritizes a fresh replica, then optional master, then a stale replica, then waiting. An acquisition already waiting with no candidates is awakened and re-evaluates when a stale fallback becomes available. Metrics expose lag under the bytes and time keys (the latter is a timedelta).

Architecture

hasql uses a composition-based architecture. Pool orchestration logic lives in BasePoolManager, while all driver-specific operations (creating pools, acquiring/releasing connections, checking master status) are encapsulated in PoolDriver implementations.

PoolDriver (ABC)                    <- driver interface (11 methods)
  ├── AiopgDriver
  ├── AiopgSaDriver
  ├── AsyncpgDriver
  │     └── AsyncpgsaDriver
  ├── Psycopg3Driver
  └── AsyncSqlAlchemyDriver

BasePoolManager (concrete)          <- has-a PoolDriver
  └── driver-specific PoolManager   <- thin wrapper: creates driver

Each driver-specific PoolManager (for example, hasql.driver.aiopg.PoolManager) is a thin subclass that passes the appropriate PoolDriver instance to BasePoolManager:

from hasql.driver.aiopg import PoolManager

# PoolManager internally creates AiopgDriver and passes it
# to BasePoolManager — no need to interact with PoolDriver directly
pool = PoolManager("postgresql://master,replica/db")

Custom drivers

You can implement a custom driver by subclassing PoolDriver:

from hasql.abc import PoolDriver
from hasql.pool_manager import BasePoolManager

class MyDriver(PoolDriver[MyPool, MyConnection]):
    # implement all abstract methods ...
    ...

pool = BasePoolManager(
    "postgresql://master,replica/db",
    driver=MyDriver(),
)

Overview

• hasql.abc.PoolDriver

  Abstract base class for database driver implementations. Each driver must implement:

  • get_pool_freesize(pool) - Return number of free connections
  • acquire_from_pool(pool, *, timeout, **kwargs) - Acquire a connection
  • release_to_pool(connection, pool, **kwargs) - Release a connection
  • is_master(connection) - Check if connection is to master
  • fetch_scalar(connection, query) - Execute a query and return one scalar
  • pool_factory(dsn, **kwargs) - Create a connection pool
  • close_pool(pool) - Gracefully close a pool
  • terminate_pool(pool) - Forcefully terminate a pool
  • is_connection_closed(connection) - Check if connection is closed
  • host(pool) - Return host address for a pool
  • pool_stats(pool) - Return PoolStats for a single pool

  Optional override:

  • prepare_pool_factory_kwargs(kwargs) - Adjust pool factory kwargs (e.g. to reserve a system connection by incrementing min/max size)

• hasql.pool_manager.BasePoolManager

  • __init__(dsn, *, driver, acquire_timeout, refresh_delay, refresh_timeout, fallback_master, master_as_replica_weight, balancer_policy, pool_factory_kwargs):

    • dsn: str - Connection string used by the connection.
    • driver: PoolDriver - Driver instance that implements database-specific pool operations. Driver-specific PoolManager classes provide this automatically.
    • acquire_timeout: Union[int, float] - Default timeout (in seconds) for connection operations. 1 sec by default.
    • refresh_delay: Union[int, float] - Delay time (in seconds) between host polls. 1 sec by default.
    • refresh_timeout: Union[int, float] - Timeout (in seconds) for trying to connect and get the host role. 30 sec by default.
    • fallback_master: bool - Use connections from master if replicas are missing. False by default.
    • master_as_replica_weight: float - Probability of using the master as a replica (from 0. to 1.; 0. - master is not used as a replica; 1. - master can be used as a replica).
    • balancer_policy: type - Connection pool balancing policy (GreedyBalancerPolicy, RandomWeightedBalancerPolicy or RoundRobinBalancerPolicy).
    • stopwatch_window_size: int - Window size for calculating the median response time of each pool.
    • pool_factory_kwargs: Optional[dict] - Connection pool creation parameters that are passed to pool factory.
    • staleness: Optional[StalenessPolicy] - Optional replica staleness policy described above.

  • coroutine async-with acquire(read_only, fallback_master, timeout, **kwargs) Acquire a connection from free pool.

    • readonly: bool - True if need return connection to replica, False - to master. False by default.
    • fallback_master: Optional[bool] - Use connections from master if replicas are missing. If None, then the default value is used.
    • master_as_replica_weight: float - Probability of using the master as a replica (from 0. to 1.).
    • timeout: Union[int, float] - Timeout (in seconds) for connection operations.
    • kwargs - Arguments to be passed to the pool acquire() method.

  • coroutine async-with acquire_master(timeout, **kwargs) Acquire a connection from free master pool. Equivalent acquire(read_only=False)

    • timeout: Union[int, float] - Timeout (in seconds) for connection operations.
    • kwargs - Arguments to be passed to the pool acquire() method.

  • coroutine async-with acquire_replica(fallback_master, timeout, **kwargs) Acquire a connection from free replica pool. Equivalent acquire(read_only=True)

    • fallback_master: Optional[bool] - Use connections from master if replicas are missing. If None, then the default value is used.
    • master_as_replica_weight: float - Probability of using the master as a replica (from 0. to 1.).
    • timeout: Union[int, float] - Timeout (in seconds) for connection operations.
    • kwargs - Arguments to be passed to the pool acquire() method.

  • coroutine close() Close pool. Mark all pool connections to be closed on getting back to pool. Closed pool doesn’t allow to acquire new connections.

  • metrics() Returns a Metrics snapshot of the entire cluster state.

  • coroutine ready(masters_count, replicas_count, timeout) Waiting for a connection to the database hosts. If masters_count is None and replicas_count is None, then connection to all hosts is expected.

    • masters_count: Optional[int] - Minimum number of master hosts. None by default.
    • replicas_count: Optional[int] - Minimum number of replica hosts. None by default.
    • timeout: Union[int, float] - Timeout for database connections. 10 seconds by default.

  • coroutine wait_masters_ready(masters_count) Waiting for connection to the specified number of database master servers.

    • masters_count: int - Minimum number of master hosts.

  • available_pool_count Property returning the total number of pools with a known role (masters + replicas).

• hasql.driver.aiopg.PoolManager (driver: AiopgDriver)

• hasql.driver.aiopg_sa.PoolManager (driver: AiopgSaDriver)

• hasql.driver.asyncpg.PoolManager (driver: AsyncpgDriver)

• hasql.driver.asyncpgsa.PoolManager (driver: AsyncpgsaDriver)

• hasql.driver.asyncsqlalchemy.PoolManager (driver: AsyncSqlAlchemyDriver)

• hasql.driver.psycopg3.PoolManager (driver: Psycopg3Driver)

The former root driver modules remain cheap compatibility import shims. hasql.psycopg3.PoolAcquireContext is an identity alias of Psycopg3AcquireContext. hasql.base retains only BasePoolManager, AbstractBalancerPolicy, TimeoutAcquireContext, and PoolAcquireContext.

Balancer policies

• hasql.balancer_policy.GreedyBalancerPolicy Chooses pool with the most free connections. If there are several such pools, a random one is taken.
• hasql.balancer_policy.RandomWeightedBalancerPolicy Selects with probabilities proportional to the inverse of each candidate's last health-check latency. Lower valid latency is more likely. If any timing is missing or invalid (None, <= 0, NaN, or infinite), all candidates receive uniform weights. The latency is health-check latency only, not query latency or pool load; selection is statistical, with no deterministic, fair, or no-starvation guarantee.
• hasql.balancer_policy.RoundRobinBalancerPolicy