hasql 0.10.0

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

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

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

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

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

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

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

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

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

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

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.

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:

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.

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

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)

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

# 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

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")

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