Source code for lcmd_db._schema

"""Runtime property schema introspection via the API."""

from __future__ import annotations

import json
import urllib.error
import urllib.request

from .exceptions import DatasetNotFoundError, DownloadError, SchemaError
from .settings import settings
from .types import EntityType, PropertyInfo




def schema(
    subset: str,
    entity_type: EntityType = EntityType.MOLECULES,
) -> list[PropertyInfo]:
    """Fetch property definitions for a subset from the API.

    Raises:
        DatasetNotFoundError: If the subset does not exist.
        DownloadError: On network / HTTP errors.
        SchemaError: If the response cannot be parsed.
    """
    url = (
        f"{settings.base_url}/api/v1/subsets/{subset}"
        f"/properties?type={entity_type.value}"
    )

    data = _fetch_json(url)
    columns = data.get("columns")
    if not isinstance(columns, list):
        raise SchemaError(f"Unexpected response shape from {url}")

    return [PropertyInfo.model_validate(col) for col in columns]



def _fetch_json(url: str) -> dict[str, object]:
    """GET a JSON endpoint, translating HTTP errors to domain exceptions."""
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=settings.timeout) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            raise DatasetNotFoundError(f"Not found: {url}") from exc
        raise DownloadError(f"HTTP {exc.code} from {url}") from exc
    except urllib.error.URLError as exc:
        raise DownloadError(f"Network error: {exc.reason}") from exc