server.web.db.dns

dns.py — generic SQLAlchemy engine factory with DNS-aware pooling.

get_dns_engine() behaves exactly like :pyfunc:`sqlalchemy.create_engine`, but ensures that every connection the pool hands out is opened against the current IP address for the hostname in your database URL. When the DNS A-record changes, stale sockets are invalidated and new ones are created transparently.

Key points:

  • Works with any synchronous SQLAlchemy dialect – no hard dependency on a particular DB-API.

  • Accepts the full argument surface of :pyfunc:`sqlalchemy.create_engine`. The function injects its own creator for DNS-aware pooling; supplying your own creator will raise a clear ValueError, because a custom creator would defeat the point of DNS-aware pooling.

  • Opens zero extra database connections while building the engine.

  • Includes clear_dns_cache() for tests.

Example:

>>> from dns_pool import get_dns_engine, clear_dns_cache
>>> engine = get_dns_engine(
...     "postgresql+psycopg2://user:pass@db.internal/mydb",
...     echo=True, pool_size=10, pool_timeout=20
... )

If the host’s IP address changes at runtime, the very next checkout from the pool raises DisconnectionError; SQLAlchemy automatically retries and a new connection is opened to the fresh IP.

Functions

clear_dns_cache(→ None)

Flush the local DNS cache and reset failure counts (useful in unit tests).

get_dns_engine(→ sqlalchemy.engine.Engine)

Create a DNS-powered engine from a SQLAlchemy URL.

get_ip_with_ttl(→ Tuple[str, int])

Get the current IP address and TTL for a hostname.

Module Contents

server.web.db.dns.clear_dns_cache() → None[source]

Flush the local DNS cache and reset failure counts (useful in unit tests).

server.web.db.dns.get_dns_engine(uri: str | sqlalchemy.engine.URL, *engine_args: Any, dns_timeout: int = 12, dns_nameservers: list[str] | None = None, dns_grace_ttl: int = 30, dns_fallback: bool = True, dns_max_retries: int = 3, dns_extend_grace: bool = True, **engine_kwargs: Any) → sqlalchemy.engine.Engine[source]

Create a DNS-powered engine from a SQLAlchemy URL.

With a normal SQLAlchemy engine, the hostname in the URL is looked up exactly once, on connection pool creation. If a DNS entry TTL expires or otherwise updates, this will prevent failover on reconnect, because the old, cached address will still be used.

This function fixes the problem by intercepting the connection factory to resolve DNS on each reconnect.

server.web.db.dns.get_ip_with_ttl(host: str, dns_timeout: int = 12, dns_nameservers: list[str] | None = None, dns_grace_ttl: int = 30, dns_max_retries: int = 3, dns_extend_grace: bool = True) → Tuple[str, int][source]

Get the current IP address and TTL for a hostname.

Parameters:

  • host (str) – The hostname to resolve

  • dns_timeout (int) – DNS query timeout in seconds

  • dns_nameservers (list[str] | None) – Custom nameservers to use

  • dns_grace_ttl (int) – Grace period for cached IPs during failures

  • dns_max_retries (int) – Maximum retry attempts for DNS resolution

  • dns_extend_grace (bool) – Whether to extend grace period on repeated failures

Returns:

A tuple of (ip_address, ttl_in_seconds)

Return type:

Tuple[str, int]