Source code for sqlspec.utils.module_loader

"""Module loading utilities for SQLSpec.

Provides functions for dynamic module imports, path resolution, and dependency
availability checking. Used for loading modules from dotted paths, converting
module paths to filesystem paths, and ensuring optional dependencies are installed.
"""

import importlib
from importlib.util import find_spec
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypeVar, cast

from sqlspec.exceptions import MissingDependencyError, SQLSpecError

if TYPE_CHECKING:
    from types import ModuleType

__all__ = (
    "OptionalDependencyFlag",
    "dependency_flag",
    "ensure_attrs",
    "ensure_cattrs",
    "ensure_fsspec",
    "ensure_litestar",
    "ensure_msgspec",
    "ensure_numpy",
    "ensure_obstore",
    "ensure_opentelemetry",
    "ensure_orjson",
    "ensure_pandas",
    "ensure_pgvector",
    "ensure_polars",
    "ensure_prometheus",
    "ensure_pyarrow",
    "ensure_pydantic",
    "ensure_uvloop",
    "import_optional",
    "import_optional_attr",
    "import_string",
    "module_available",
    "module_to_os_path",
    "reset_dependency_cache",
    "resolve_optional_attr",
)


# =============================================================================
# Dependency Availability Checking
# =============================================================================

_dependency_cache: "dict[str, bool]" = {}
_optional_module_cache: "dict[str, ModuleType | None]" = {}
T = TypeVar("T")



[docs]
def module_available(module_name: str) -> bool:
    """Return True if the given module can be resolved.

    The result is cached per interpreter session. Call
    :func:`reset_dependency_cache` to invalidate cached entries when
    tests manipulate ``sys.path``.

    Args:
        module_name: Dotted module path to check.

    Returns:
        True if :mod:`importlib` can find the module, False otherwise.
    """

    cached = _dependency_cache.get(module_name)
    if cached is not None:
        return cached

    try:
        is_available = find_spec(module_name) is not None
    except ModuleNotFoundError:
        is_available = False

    _dependency_cache[module_name] = is_available
    return is_available

[docs]

def reset_dependency_cache(module_name: str | None = None) -> None: """Clear cached availability for one module or the entire cache. Args: module_name: Specific dotted module path to drop from the cache. Clears the full cache when ``None``. """ if module_name is None: _dependency_cache.clear() _optional_module_cache.clear() return _dependency_cache.pop(module_name, None) _optional_module_cache.pop(module_name, None)