from .connectivity import (
build_db_uri,
check_db_connection
)
[docs]
def check_mlflow_connection(tracking_uri: str, timeout: int = 5) -> bool:
"""Wrapper to check MLflow connection from a raw URI."""
# Simple parse to convert string URI to dict-like object for the checker
from urllib.parse import urlparse
parsed = urlparse(tracking_uri)
cfg = {
'driver': parsed.scheme,
'host': parsed.hostname,
'port': parsed.port,
'user': parsed.username,
'password': parsed.password,
'db_name': parsed.path.lstrip('/')
}
return check_db_connection(cfg, timeout)
[docs]
def check_optuna_connection(storage_url: str, timeout: int = 5) -> bool:
"""Wrapper to check Optuna connection from a raw URI."""
# Reuse logic
return check_mlflow_connection(storage_url, timeout)