from __future__ import annotations
import importlib
import sys
import warnings
from typing import Iterable
def _warn_alias(alias: str, target: str) -> None:
warnings.warn(
f"`{alias}` is a compatibility alias for `{target}`; prefer `{target}` in new code. "
"The legacy `briefcase` namespace will be removed in 2.1.31.",
DeprecationWarning,
stacklevel=3,
)
def _import_with_guidance(module_path: str, alias_name: str):
try:
return importlib.import_module(module_path)
except ModuleNotFoundError as exc:
missing = exc.name or "unknown dependency"
raise ImportError(
f"Compatibility alias `{alias_name}` could not import `{module_path}` "
f"because dependency `{missing}` is missing. Install required optional "
"dependencies for this feature or import from supported core APIs."
) from exc
def _bridge_module(current_module: str, legacy_module: str, submodules: Iterable[str] = ()) -> None:
legacy = _import_with_guidance(legacy_module, current_module)
module = sys.modules[current_module]
module.__doc__ = getattr(legacy, "__doc__", module.__doc__)
legacy_all = list(getattr(legacy, "__all__", []))
def __getattr__(name: str):
try:
return getattr(legacy, name)
except AttributeError:
submodule_path = f"{legacy_module}.{name}"
shim_path = f"{current_module}.{name}"
_warn_alias(shim_path, submodule_path)
imported = _import_with_guidance(submodule_path, shim_path)
sys.modules[shim_path] = imported
setattr(module, name, imported)
return imported
def __dir__():
return sorted(set(dir(legacy)) | set(module.__dict__.keys()))
module.__getattr__ = __getattr__ module.__dir__ = __dir__
exports = legacy_all or [name for name in dir(legacy) if not name.startswith("_")]
module.__all__ = list(exports)
for name in exports:
try:
setattr(module, name, getattr(legacy, name))
except Exception:
continue
for submodule in submodules:
legacy_sub = f"{legacy_module}.{submodule}"
shim_sub = f"{current_module}.{submodule}"
try:
imported = importlib.import_module(legacy_sub)
except Exception:
continue
sys.modules[shim_sub] = imported
setattr(module, submodule, imported)
def versioned(*args, **kwargs):
_warn_alias("briefcase_ai.versioned", "briefcase.integrations.lakefs.versioned")
impl = _import_with_guidance(
"briefcase.integrations.lakefs.decorators", "briefcase_ai.versioned"
)
return impl.versioned(*args, **kwargs)
def lakefs_versioned(*args, **kwargs):
_warn_alias(
"briefcase_ai.lakefs_versioned", "briefcase.integrations.lakefs.lakefs_versioned"
)
impl = _import_with_guidance(
"briefcase.integrations.lakefs.decorators", "briefcase_ai.lakefs_versioned"
)
return impl.lakefs_versioned(*args, **kwargs)
def versioned_context(*args, **kwargs):
_warn_alias(
"briefcase_ai.versioned_context", "briefcase.integrations.lakefs.versioned_context"
)
impl = _import_with_guidance(
"briefcase.integrations.lakefs.context", "briefcase_ai.versioned_context"
)
return impl.versioned_context(*args, **kwargs)
def lakefs_context(*args, **kwargs):
_warn_alias("briefcase_ai.lakefs_context", "briefcase.integrations.lakefs.lakefs_context")
impl = _import_with_guidance(
"briefcase.integrations.lakefs.context", "briefcase_ai.lakefs_context"
)
return impl.lakefs_context(*args, **kwargs)
def briefcase_workflow(*args, **kwargs):
_warn_alias("briefcase_ai.briefcase_workflow", "briefcase.correlation.briefcase_workflow")
impl = _import_with_guidance("briefcase.correlation.workflow", "briefcase_ai.briefcase_workflow")
return impl.briefcase_workflow(*args, **kwargs)