Project-SDE-WP-Suite/server/assets_db.py

"""Read-only reader for the Micron asset catalog.

The work package creator used to ask people to paste a controls.dev link for
every asset. Assets actually live in the Micron database — a SQL Server instance
that is NOT part of this repo and whose schema is not managed here. This module
gives the API a *read-only* window onto it so the creator can offer a searchable
picker instead of free-text links.

How it works: the whole catalog is fetched in one query and handed to the browser
when the creator loads. Searching then happens in the browser with no round trip
at all. The catalog is a list of asset IDs — about 9k of them today and not
expected past 100k — so it is small enough to send whole, and it is slow-moving
reference data, so there is nothing to gain from querying it per keystroke and a
lot of latency to lose. A short server-side cache keeps a room full of people
opening the page from turning into a query each.

Other deliberate constraints:

  * **Read-only, always.** The only statement in this file is the SELECT below.
    Point it at a login with `db_datareader` and nothing else.
  * **No prime_db dependency.** A plain SQLAlchemy connection built from a
    connection string, kept separate from the app's own engine in `db.py`, so a
    Micron outage can never affect the suite's own database.

Unconfigured is a first-class state: with no `MICRON_DB_URL` set, `configured()`
returns False, the API says so, and the UI falls back to manual entry. The suite
boots and runs fine without the Micron database being reachable.
"""
import os
import time
import logging
import threading

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError

log = logging.getLogger(__name__)

try:
    from dotenv import load_dotenv
    load_dotenv()
except Exception:
    pass


# ── The query ─────────────────────────────────────────────────────────────────
# The only place the Micron schema appears; everything else here is plumbing.
# Returns one row per asset, aliased `tag`. No row cap: the catalog is small
# enough to hand over whole, and a partial list would silently hide assets.
#
# Add a WHERE clause here if some rows should never be offered at all
# (decommissioned assets, other sites, …). Filtering at the source keeps the
# payload small, which matters more than anything else here.
ASSET_QUERY = """
    SELECT a.AssetID AS tag
      FROM Asset.Asset AS a
     ORDER BY a.AssetID
"""

def _env_int(name: str, default: int) -> int:
    """A malformed tuning knob degrades to its default; it must never keep the
    suite from booting. app.py imports this module unconditionally, so a bare
    int() here would turn "300s" in someone's .env into a crash-looping API -
    the total-outage switch an OPTIONAL feature is not allowed to own."""
    raw = os.getenv(name, "")
    try:
        return int(raw.strip()) if raw.strip() else default
    except ValueError:
        log.warning("%s=%r is not an integer; using the default %d.", name, raw, default)
        return default


# How long a fetched catalog is reused before the next page load re-queries.
CACHE_SECONDS = _env_int("MICRON_ASSETS_CACHE_SECONDS", 300)    # 5 min
# How long a FAILURE is remembered before the next request retries the source.
# Without this, every page load during a Micron outage spends CONNECT_TIMEOUT
# seconds inside a worker thread; enough concurrent loads exhaust the app's
# shared sync threadpool and take unrelated endpoints down with the picker.
FAIL_CACHE_SECONDS = _env_int("MICRON_ASSETS_FAIL_CACHE_SECONDS", 30)
CONNECT_TIMEOUT = 8


class AssetSourceError(RuntimeError):
    """The catalog is configured but could not be read."""


# ── Engine (lazy, process-wide) ───────────────────────────────────────────────
# A full SQLAlchemy URL, e.g.
#   mssql+pymssql://user:pass@host:1433/MicronDB
#   mssql+pyodbc://user:pass@host/MicronDB?driver=ODBC+Driver+18+for+SQL+Server
# URL-encode any special characters in the password.
_engine = None
_engine_lock = threading.Lock()


def _db_url() -> str:
    return os.getenv("MICRON_DB_URL", "").strip()


def configured() -> bool:
    return bool(_db_url())


def _validate_url(url: str) -> None:
    """Catch the one URL mistake that produces a baffling error message.

    A password containing an unencoded '@' makes the URL ambiguous: the parser
    splits on the first '@', so part of the password ends up parsed as the host.
    The driver then reports a connection failure against a nonsense hostname that
    happens to contain a fragment of the password — confusing to read and unsafe
    to display. Detect it here and say plainly what is wrong.

    Nothing from the URL is included in the message; it never leaves this process.
    """
    authority = url.split("://", 1)[-1].split("/", 1)[0]
    if authority.count("@") > 1:
        raise AssetSourceError(
            "MICRON_DB_URL is ambiguous: the username or password contains an "
            "unencoded '@'. Percent-encode the special characters — @ = %40, "
            ": = %3A, / = %2F, # = %23, ? = %3F, %  = %25."
        )


def _connect_args(url: str) -> dict:
    """Per-driver connect timeouts, so an unreachable Micron host fails fast
    instead of tying up a worker until the OS gives up."""
    if url.startswith("mssql+pymssql"):
        return {"login_timeout": CONNECT_TIMEOUT, "timeout": CONNECT_TIMEOUT}
    if url.startswith("mssql+pyodbc"):
        return {"timeout": CONNECT_TIMEOUT}
    return {}


def _get_engine():
    global _engine
    if _engine is not None:
        return _engine
    url = _db_url()
    if not url:
        raise AssetSourceError("The Micron DB is not configured.")
    _validate_url(url)
    with _engine_lock:
        if _engine is None:
            try:
                _engine = create_engine(
                    url,
                    connect_args=_connect_args(url),
                    pool_pre_ping=True,   # a recycled dead connection retries instead of erroring
                    pool_recycle=1800,
                    pool_size=1,          # one catalog query now and then, not a workload
                    max_overflow=1,
                    future=True,
                )
            except Exception as exc:      # bad URL, missing driver package, …
                # See the note on load() — the exception text can echo the
                # connection string, so it is logged and not propagated.
                log.error("Micron asset catalog: could not open the connection: %s", exc)
                raise AssetSourceError(
                    "Could not open a connection to the Micron DB. "
                    "Check MICRON_DB_URL and the API log for the driver error."
                ) from exc
    return _engine


# ── Cache ─────────────────────────────────────────────────────────────────────
# Every page load asks for the whole catalog, so without this a shift change
# would be one full-table query per person. Held per worker process.
_cache: list[dict] | None = None
_cached_at = 0.0
_error: str | None = None      # negative cache: the last failure's user-safe text
_error_at = 0.0
_cache_lock = threading.Lock()


def load(force: bool = False) -> list[dict]:
    """Return the whole catalog as [{'tag': …}, …]. Never writes.

    Failures are handled in two tiers so a Micron outage stays the picker's
    problem and never the suite's (the module contract above):
      * a previously fetched catalog is served STALE - it is slow-moving
        reference data, and old-but-real beats an error;
      * with nothing to serve, the failure itself is cached for
        FAIL_CACHE_SECONDS, so an outage costs one CONNECT_TIMEOUT per window
        instead of one per page load stacking up in the shared threadpool."""
    global _cache, _cached_at, _error, _error_at
    with _cache_lock:
        if _cache is not None and not force and (time.monotonic() - _cached_at) < CACHE_SECONDS:
            return _cache
        if (_error is not None and not force
                and (time.monotonic() - _error_at) < FAIL_CACHE_SECONDS
                and _cache is None):
            raise AssetSourceError(_error)

    try:
        engine = _get_engine()
        with engine.connect() as conn:
            result = conn.execute(text(ASSET_QUERY)).mappings().all()
    except AssetSourceError as exc:
        # _get_engine already logged and sanitised; remember or stale-serve.
        with _cache_lock:
            if _cache is not None:
                log.warning("Micron asset catalog unavailable; serving the cached "
                            "catalog (%d rows).", len(_cache))
                return _cache
            _error, _error_at = str(exc), time.monotonic()
        raise
    except SQLAlchemyError as exc:
        # The driver's message is NOT propagated. AssetSourceError text reaches the
        # browser, and connection errors quote the host, the login, and — when the
        # URL is malformed — fragments of the password. Operators get the detail
        # from the API log, where it belongs; users get a message they can act on.
        log.error("Micron asset catalog query failed: %s", exc)
        msg = ("The Micron DB could not be read. Check that the host is "
               "reachable, that the login has SELECT on the asset table, and that "
               "ASSET_QUERY matches the real schema — the API log has the driver error.")
        with _cache_lock:
            if _cache is not None:
                log.warning("Micron asset catalog unavailable; serving the cached "
                            "catalog (%d rows).", len(_cache))
                return _cache
            _error, _error_at = msg, time.monotonic()
        raise AssetSourceError(msg) from exc

    # Drop rows with no identifier — an asset with no tag is not selectable and
    # would render as a blank line in the picker.
    rows = [{"tag": str(r["tag"])} for r in result if r.get("tag") not in (None, "")]
    with _cache_lock:
        _cache, _cached_at = rows, time.monotonic()
        _error = None
    return rows


def status() -> dict:
    """Describe the source for the UI, so it can explain itself rather than just
    showing an empty dropdown."""
    if not configured():
        return {
            "configured": False, "ok": False, "count": 0,
            "detail": "The Micron DB is not configured — enter assets manually.",
        }
    try:
        rows = load()
    except AssetSourceError as exc:
        return {"configured": True, "ok": False, "count": 0, "detail": str(exc)}
    return {"configured": True, "ok": True, "count": len(rows),
            "detail": f"{len(rows):,} asset IDs from the Micron DB."}