server.web.db.dns
=================

.. py:module:: server.web.db.dns

.. autoapi-nested-parse::

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

.. autoapisummary::

   server.web.db.dns.clear_dns_cache
   server.web.db.dns.get_dns_engine
   server.web.db.dns.get_ip_with_ttl


Module Contents
---------------

.. py:function:: clear_dns_cache() -> None

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


.. py:function:: 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

   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.


.. py:function:: 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]

   Get the current IP address and TTL for a hostname.

   :param host: The hostname to resolve
   :type host: str
   :param dns_timeout: DNS query timeout in seconds
   :type dns_timeout: int
   :param dns_nameservers: Custom nameservers to use
   :type dns_nameservers: list[str] | None
   :param dns_grace_ttl: Grace period for cached IPs during failures
   :type dns_grace_ttl: int
   :param dns_max_retries: Maximum retry attempts for DNS resolution
   :type dns_max_retries: int
   :param dns_extend_grace: Whether to extend grace period on repeated failures
   :type dns_extend_grace: bool

   :returns: A tuple of (ip_address, ttl_in_seconds)
   :rtype: Tuple[str, int]