Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
### Under the Hood

- Document serverless environment configuration for Python models (thanks @TangoEnSkai!) ([#1649](https://github.com/databricks/dbt-databricks/pull/1649) resolves [#1055](https://github.com/databricks/dbt-databricks/issues/1055))
- Normalize artifact URLs as well as source registries in `check_uv_lock_public_urls.py --fix`, so regenerating `uv.lock` against a mirrored index no longer needs a manual edit (tooling-only, no runtime impact) ([#1675](https://github.com/databricks/dbt-databricks/pull/1675))

## dbt-databricks 1.12.5 (Sep 1, 2026)

Expand Down
37 changes: 28 additions & 9 deletions scripts/check_uv_lock_public_urls.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,9 +10,14 @@

PUBLIC_REGISTRY = "https://pypi.org/simple"
PUBLIC_PACKAGE_URL_PREFIX = "https://files.pythonhosted.org/packages/"
PUBLIC_PACKAGE_HOST = PUBLIC_PACKAGE_URL_PREFIX.removesuffix("/packages/")
REGISTRY_PATTERN = re.compile(r'registry = "([^"]+)"')
SOURCE_REGISTRY_PATTERN = re.compile(r'(source = \{ registry = ")([^"]+)(" \})')
URL_PATTERN = re.compile(r'url = "([^"]+)"')
# A mirror serves the same artifact tree, so only the host differs: the /packages/ path and
# the hash recorded next to it are identical to public PyPI's. Anything not shaped like that
# is left alone for the check to report rather than rewritten on a guess.
MIRROR_PACKAGE_URL_PATTERN = re.compile(r'(url = ")https?://[^"/]+(/packages/[^"]+")')


def check_uv_lock(lockfile: Path) -> list[str]:
Expand All @@ -30,29 +35,39 @@ def check_uv_lock(lockfile: Path) -> list[str]:
return failures


def fix_uv_lock(lockfile: Path) -> int:
def fix_uv_lock(lockfile: Path) -> tuple[int, int]:
contents = lockfile.read_text()
change_count = 0
registry_count = 0
url_count = 0

def normalize_registry(match: re.Match[str]) -> str:
nonlocal change_count
nonlocal registry_count
if match.group(2) == PUBLIC_REGISTRY:
return match.group(0)
change_count += 1
registry_count += 1
return f"{match.group(1)}{PUBLIC_REGISTRY}{match.group(3)}"

def normalize_package_url(match: re.Match[str]) -> str:
nonlocal url_count
rewritten = f"{match.group(1)}{PUBLIC_PACKAGE_HOST}{match.group(2)}"
if rewritten == match.group(0):
return match.group(0)
url_count += 1
return rewritten

normalized = SOURCE_REGISTRY_PATTERN.sub(normalize_registry, contents)
if change_count:
normalized = MIRROR_PACKAGE_URL_PATTERN.sub(normalize_package_url, normalized)
if registry_count or url_count:
lockfile.write_text(normalized)
return change_count
return registry_count, url_count


def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--fix",
action="store_true",
help="normalize source registries before checking the lock file",
help="normalize source registries and package URLs before checking the lock file",
)
parser.add_argument("lockfile", nargs="?", type=Path, default=Path("uv.lock"))
args = parser.parse_args()
Expand All @@ -63,8 +78,12 @@ def main() -> int:
return 1

if args.fix:
change_count = fix_uv_lock(lockfile)
print(f"Normalized {change_count} source registry entries in {lockfile}", flush=True)
registry_count, url_count = fix_uv_lock(lockfile)
print(
f"Normalized {registry_count} source registry entries and "
f"{url_count} package URLs in {lockfile}",
flush=True,
)

failures = check_uv_lock(lockfile)
if failures:
Expand Down
90 changes: 90 additions & 0 deletions tests/unit/test_check_uv_lock_public_urls.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
import importlib.util
import sys
from pathlib import Path

_SCRIPTS_DIR = Path(__file__).resolve().parents[2] / "scripts"
if str(_SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(_SCRIPTS_DIR))

_SCRIPT = _SCRIPTS_DIR / "check_uv_lock_public_urls.py"
_spec = importlib.util.spec_from_file_location("check_uv_lock_public_urls", _SCRIPT)
assert _spec is not None and _spec.loader is not None
checker = importlib.util.module_from_spec(_spec)
sys.modules["check_uv_lock_public_urls"] = checker
_spec.loader.exec_module(checker)

MIRROR = "https://pypi-proxy.example.com"
PACKAGE_PATH = "/packages/29/77/6f5df1c68bf/agate-1.9.1.tar.gz"

PUBLIC_LOCK = f"""\
[[package]]
name = "agate"
version = "1.9.1"
source = {{ registry = "https://pypi.org/simple" }}
sdist = {{ url = "https://files.pythonhosted.org{PACKAGE_PATH}", hash = "sha256:abc" }}
"""

MIRROR_LOCK = f"""\
[[package]]
name = "agate"
version = "1.9.1"
source = {{ registry = "{MIRROR}/simple" }}
sdist = {{ url = "{MIRROR}{PACKAGE_PATH}", hash = "sha256:abc" }}
"""


class TestCheckUvLock:
def test_public_lock_passes(self, tmp_path):
lockfile = tmp_path / "uv.lock"
lockfile.write_text(PUBLIC_LOCK)
assert checker.check_uv_lock(lockfile) == []

def test_mirror_registry_and_url_both_reported(self, tmp_path):
lockfile = tmp_path / "uv.lock"
lockfile.write_text(MIRROR_LOCK)
failures = checker.check_uv_lock(lockfile)
assert len(failures) == 2
assert any("registry URL is not public PyPI" in failure for failure in failures)
assert any("package URL is not public PyPI" in failure for failure in failures)


class TestFixUvLock:
def test_fix_normalizes_registry_and_package_url(self, tmp_path):
lockfile = tmp_path / "uv.lock"
lockfile.write_text(MIRROR_LOCK)

registry_count, url_count = checker.fix_uv_lock(lockfile)

assert (registry_count, url_count) == (1, 1)
assert lockfile.read_text() == PUBLIC_LOCK
assert checker.check_uv_lock(lockfile) == []

def test_fix_is_a_no_op_on_a_public_lock(self, tmp_path):
lockfile = tmp_path / "uv.lock"
lockfile.write_text(PUBLIC_LOCK)

assert checker.fix_uv_lock(lockfile) == (0, 0)
assert lockfile.read_text() == PUBLIC_LOCK

def test_fix_preserves_the_artifact_path_and_hash(self, tmp_path):
"""Only the host may change: the /packages/ path identifies the artifact and the hash
next to it is what actually guarantees integrity."""
lockfile = tmp_path / "uv.lock"
lockfile.write_text(MIRROR_LOCK)

checker.fix_uv_lock(lockfile)

contents = lockfile.read_text()
assert PACKAGE_PATH in contents
assert 'hash = "sha256:abc"' in contents

def test_fix_leaves_urls_that_are_not_mirror_shaped(self, tmp_path):
"""A URL without a /packages/ path is not a mirror of the public artifact tree, so
rewriting its host would be a guess. Leave it for the check to report."""
lockfile = tmp_path / "uv.lock"
original = 'sdist = { url = "https://example.com/downloads/agate-1.9.1.tar.gz" }\n'
lockfile.write_text(original)

assert checker.fix_uv_lock(lockfile) == (0, 0)
assert lockfile.read_text() == original
assert len(checker.check_uv_lock(lockfile)) == 1