Registry - simplify versioning

mindtrace/registry/mindtrace/registry/core/_registry_core.py

@@ -18,6 +18,7 @@
     RegistryVersionConflict,
 )
 from mindtrace.registry.core.types import (
+    _POP_MISSING,
     ERROR_UNKNOWN,
     VERSION_PENDING,
     BatchResult,
@@ -803,14 +804,24 @@ def _delete_single(
         if version is None:
             # Delete all versions
             if not self.version_objects:
+                if not self.has_object(name, "1"):
+                    raise RegistryObjectNotFound(f"Object {name} does not exist")
                 versions_to_delete = ["1"]
             else:
                 versions_to_delete = self.list_versions(name)
                 if not versions_to_delete:
                     raise RegistryObjectNotFound(f"Object {name} does not exist")
+        elif version == "latest":
+            # Resolve "latest" to concrete version
+            latest = self._latest(name)
+            if latest is None:
+                raise RegistryObjectNotFound(f"Object {name}@latest does not exist")
+            versions_to_delete = [latest]
         else:
             # Explicit version only — validate format, pass directly
             validated = self._validate_version(version)
+            if not self.has_object(name, validated):
+                raise RegistryObjectNotFound(f"Object {name}@{validated} not exist")
             versions_to_delete = [validated]
 
         # Delete and check for errors
@@ -861,6 +872,17 @@ def _delete_batch(
                     for ver in all_versions:
                         items_to_delete.append((n, ver))
                         resolved_to_original[(n, ver)] = original_key
+            elif v == "latest":
+                # Delete latest version
+                latest = self._latest(n)
+                if latest is None:
+                    result.errors[original_key] = {
+                        "error": "RegistryObjectNotFound",
+                        "message": f"Object {n} does not exist",
+                    }
+                else:
+                    items_to_delete.append((n, latest))
+                    resolved_to_original[(n, latest)] = original_key
             else:
                 # Explicit version only — validate format, pass directly
                 try:
@@ -934,7 +956,7 @@ def info(self, name: str | None = None, version: str | None = None) -> Dict[str,
             items = [(n, v) for n in self.list_objects() for v in self.list_versions(n)]
         elif version is not None:
             # Specific version (resolve "latest")
-            resolved_version = self._latest(name) if version == "latest" else version
+            resolved_version = self._latest(name) if version == "latest" else self._validate_version(version)
             items = [(name, resolved_version)] if resolved_version else []
         else:
             # All versions for one object
@@ -1186,13 +1208,20 @@ def _validate_version(self, version: str | None) -> str:
         # Remove any 'v' prefix
         if version.startswith("v"):
             version = version[1:]
+        # if more than 3 components, raise error
+        if len(version.split(".")) > 3:
+            raise ValueError(
+                f"Invalid version string '{version}'. Must be in semantic versioning format (e.g. '1', '1.0', '1.0.0')"
+            )
 
         # Split into components and validate
         try:
             components = version.split(".")
-            # Convert each component to int to validate
-            [int(c) for c in components]
-            return version
+            int_components = [int(c) for c in components]
+            # Strip trailing zeros: "1.0.0" → "1", "1.1.0" → "1.1"
+            while len(int_components) > 1 and int_components[-1] == 0:
+                int_components.pop()
+            return ".".join(str(c) for c in int_components)
         except ValueError:
             raise ValueError(
                 f"Invalid version string '{version}'. Must be in semantic versioning format (e.g. '1', '1.0', '1.0.0')"
@@ -1589,9 +1618,10 @@ def clear(self, clear_registry_metadata: bool = False) -> None:
             except Exception as e:
                 self.logger.warning(f"Could not clear registry metadata: {e}")
 
-    def pop(self, key: str, default: Any = None) -> Any:
+    def pop(self, key: str, default: Any = _POP_MISSING) -> Any:
         """Remove and return an object from the registry.
 
+
         Args:
             key: The object name, optionally including version (e.g. "name@version")
             default: The value to return if the object doesn't exist
@@ -1602,29 +1632,20 @@ def pop(self, key: str, default: Any = None) -> Any:
         Raises:
             KeyError: If the object doesn't exist and no default is provided.
         """
+        name, parsed_version = self._parse_key(key)
+        requested_version = parsed_version or "latest"
         try:
-            name, version = self._parse_key(key)
-            if version is None:
-                version = self._latest(name)
-                if version is None:
-                    if default is not None:
-                        return default
+            value = self.load(name=name, version=requested_version)
+        except (RegistryObjectNotFound, ValueError):
+            if default is _POP_MISSING:
+                if parsed_version is None:
                     raise KeyError(f"Object {name} does not exist")
+                raise KeyError(f"Object {name} version {parsed_version} does not exist")
+            return default
 
-            # Check existence first
-            if not self.has_object(name, version):
-                if default is not None:
-                    return default
-                raise KeyError(f"Object {name} version {version} does not exist")
-
-            # Load and delete (backend handles locking internally)
-            value = self.load(name=name, version=version)
-            self.delete(name=name, version=version)
-            return value
-        except KeyError:
-            if default is not None:
-                return default
-            raise
+        delete_version = parsed_version if parsed_version is not None else self._latest(name)
+        self.delete(name=name, version=delete_version)
+        return value
 
     def setdefault(self, key: str, default: Any = None) -> Any:
         """Get an object from the registry, setting it to default if it doesn't exist.

mindtrace/registry/mindtrace/registry/core/registry.py

@@ -16,7 +16,7 @@
 from mindtrace.registry.backends.registry_backend import RegistryBackend
 from mindtrace.registry.core._registry_core import _RegistryCore
 from mindtrace.registry.core.exceptions import RegistryObjectNotFound
-from mindtrace.registry.core.types import BatchResult, OnConflict, VerifyLevel
+from mindtrace.registry.core.types import _POP_MISSING, BatchResult, OnConflict, VerifyLevel
 
 
 class Registry(Mindtrace):
@@ -193,13 +193,16 @@ def _get_cache_dir(backend_uri: str | Path, config: Dict[str, Any] | None = None
         temp_dir = Path(config["MINDTRACE_DIR_PATHS"]["TEMP_DIR"]).expanduser().resolve()
         return temp_dir / f"registry_cache_{uri_hash}"
 
-    def _is_cache_stale(self, name: str, version: str | None) -> bool:
-        """Check if a cached item is stale by comparing hashes with remote."""
-        try:
-            resolved_version = version if version and version != "latest" else self._remote._latest(name)
-            if not resolved_version:
-                return True
+    @staticmethod
+    def _hashes_indicate_stale(remote_hash: str | None, cache_hash: str | None) -> bool:
+        """Return True when cache entry should be treated as stale."""
+        if not remote_hash or not cache_hash:
+            return True
+        return remote_hash != cache_hash
 
+    def _is_cache_stale(self, name: str, resolved_version: str) -> bool:
+        """Check if a cached item is stale for a resolved version."""
+        try:
             try:
                 remote_meta = self._remote.backend.fetch_metadata(name, resolved_version).first()
             except Exception:
@@ -213,14 +216,9 @@ def _is_cache_stale(self, name: str, version: str | None) -> bool:
             remote_hash = remote_meta.metadata.get("hash") if remote_meta and remote_meta.ok else None
             cache_hash = cache_meta.metadata.get("hash") if cache_meta and cache_meta.ok else None
 
-            if not remote_hash:
-                return True
-            if not cache_hash:
-                return True
-
-            return remote_hash != cache_hash
+            return self._hashes_indicate_stale(remote_hash, cache_hash)
         except Exception as e:
-            self.logger.debug(f"Error checking cache staleness for {name}@{version}: {e}")
+            self.logger.debug(f"Error checking cache staleness for {name}@{resolved_version}: {e}")
             return True
 
     def _find_stale_indices(self, resolved: List[tuple[str, str]], indices: List[int]) -> set[int]:
@@ -242,10 +240,7 @@ def _find_stale_indices(self, resolved: List[tuple[str, str]], indices: List[int
             remote_hash = remote_meta.metadata.get("hash") if remote_meta and remote_meta.ok else None
             cache_hash = cache_meta.metadata.get("hash") if cache_meta and cache_meta.ok else None
 
-            # If we can't verify either side, treat cache as stale (consistent with _is_cache_stale).
-            if not remote_hash or not cache_hash:
-                stale.add(i)
-            elif remote_hash != cache_hash:
+            if self._hashes_indicate_stale(remote_hash, cache_hash):
                 stale.add(i)
 
         return stale
@@ -404,7 +399,7 @@ def _load_single_cached(
         **kwargs,
     ) -> Any:
         """Load a single object with cache-first pattern."""
-        resolved_v = version if version and version != "latest" else self._remote._latest(name)
+        resolved_v = self._remote._resolve_load_version(name, version)
         check_staleness = verify == VerifyLevel.FULL
 
         # Try cache first
@@ -563,23 +558,7 @@ def delete(
 
         # Delete from cache (best effort)
         try:
-            names_list = name if isinstance(name, list) else [name]
-            versions_list = version if isinstance(version, list) else [version] * len(names_list)
-
-            for n, v in zip(names_list, versions_list):
-                try:
-                    if v is None:
-                        for ver in self._cache.list_versions(n):
-                            try:
-                                self._cache.delete(n, ver)
-                            except Exception:
-                                pass
-                    else:
-                        resolved_v = v if v != "latest" else self._cache._latest(n)
-                        if resolved_v and self._cache.has_object(n, resolved_v):
-                            self._cache.delete(n, resolved_v)
-                except Exception:
-                    pass
+            self._cache.delete(name, version)
         except Exception as e:
             self.logger.warning(f"Error deleting from cache: {e}")
 
@@ -645,15 +624,7 @@ def __delitem__(self, key: str | list[str]) -> None:
             self.delete(name=names, version=versions)
             return
         try:
-            name, version = names[0], versions[0]
-            if version is None:
-                if not self._core.list_versions(name):
-                    raise RegistryObjectNotFound(f"Object {name} does not exist")
-            else:
-                exists = self._core.backend.has_object([name], [version])
-                if not exists.get((name, version), False):
-                    raise RegistryObjectNotFound(f"Object {name}@{version} does not exist")
-            self.delete(name, version)
+            self.delete(names[0], versions[0])
         except (ValueError, RegistryObjectNotFound) as e:
             raise KeyError(f"Object not found: {key}") from e
 
@@ -683,28 +654,19 @@ def values(self) -> List[Any]:
     def items(self) -> List[tuple[str, Any]]:
         return [(name, self[name]) for name in self.keys()]
 
-    def pop(self, key: str, default: Any = None) -> Any:
+    def pop(self, key: str, default: Any = _POP_MISSING) -> Any:
+        if not self._cached:
+            return self._core.pop(key, default)
+
+        value = self._remote.pop(key, default)
+
         try:
             name, version = self._core._parse_key(key)
-            if version is None:
-                version = self._core._latest(name)
-                if version is None:
-                    if default is not None:
-                        return default
-                    raise KeyError(f"Object {name} does not exist")
-
-            if not self._core.has_object(name, version):
-                if default is not None:
-                    return default
-                raise KeyError(f"Object {name} version {version} does not exist")
-
-            value = self.load(name=name, version=version)
-            self.delete(name=name, version=version)
-            return value
-        except KeyError:
-            if default is not None:
-                return default
-            raise
+            self._cache.delete(name=name, version=version)
+        except Exception:
+            pass
+
+        return value
 
     def setdefault(self, key: str, default: Any = None) -> Any:
         try:

mindtrace/registry/mindtrace/registry/core/types.py

@@ -15,7 +15,7 @@
 
 VERSION_PENDING = "pending"  # Placeholder for version not yet assigned
 ERROR_UNKNOWN = "UnknownError"  # Fallback error type when error info unavailable
-
+_POP_MISSING = object()
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Enums

tests/integration/mindtrace/registry/backend/test_gcp_registry_backend_integration.py

@@ -651,9 +651,9 @@ def test_version_objects_keeps_history(gcp_registry):
 
     # List versions should show all
     versions = gcp_registry.list_versions("test:history")
-    assert "1.0.0" in versions
-    assert "2.0.0" in versions
-    assert "3.0.0" in versions
+    assert "1" in versions
+    assert "2" in versions
+    assert "3" in versions
 
 
 def test_unversioned_registry_single_version(unversioned_registry):
@@ -747,8 +747,8 @@ def test_registry_version_listing(gcp_registry):
     gcp_registry.save("test:versioned", "v2", version="2.0.0")
 
     versions = gcp_registry.list_versions("test:versioned")
-    assert "1.0.0" in versions
-    assert "2.0.0" in versions
+    assert "1" in versions
+    assert "2" in versions
 
 
 def test_concurrent_save_load(gcp_registry):

tests/integration/mindtrace/registry/core/test_all_supported_backends.py

@@ -174,9 +174,9 @@ def test_versioning(registry, test_config):
     # List versions
     versions = registry.list_versions("test:versioning")
     assert len(versions) == 3
-    assert "1.0.0" in versions
+    assert "1" in versions
     assert "1.0.1" in versions
-    assert "1.1.0" in versions
+    assert "1.1" in versions
 
     # Load specific version
     loaded_config = registry.load("test:versioning", version="1.0.1")
@@ -219,8 +219,8 @@ def test_info(registry, test_config):
 
     # Get info for all versions - returns dict of {version: metadata}
     all_info = registry.info("test:info")
-    assert "1.0.0" in all_info
-    assert all_info["1.0.0"]["metadata"]["description"] == "test object"
+    assert "1" in all_info
+    assert all_info["1"]["metadata"]["description"] == "test object"
 
 
 @pytest.mark.slow
@@ -622,7 +622,7 @@ def test_object_discovery(registry):
     versions = registry.list_versions(f"{test_prefix}object:1")
     assert len(versions) == 2
     assert "1" in versions  # Auto-generated version
-    assert "2.0.0" in versions
+    assert "2" in versions
 
 
 def test_metadata_operations(registry):
@@ -876,9 +876,9 @@ def test_versioned_registry_keeps_history(registry):
 
     # List versions should show all
     versions = registry.list_versions("test:history")
-    assert "1.0.0" in versions
-    assert "2.0.0" in versions
-    assert "3.0.0" in versions
+    assert "1" in versions
+    assert "2" in versions
+    assert "3" in versions
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -918,17 +918,12 @@ def test_delete_all_versions(registry):
 
 
 def test_delete_nonexistent_raises(registry):
-    """Test that deleting nonexistent object raises error when version is None.
-    Note: Deleting with a specific version is idempotent (succeeds even if not found).
-    """
+    """Test that deleting a single nonexistent object raises error."""
     from mindtrace.registry.core.exceptions import RegistryObjectNotFound
 
-    # Deleting with specific version is idempotent - no error
-    registry.delete("test:nonexistent", version="1.0.0")  # Should not raise
-
-    # Deleting with version=None raises because object has no versions
+    # Deleting a single nonexistent object raises error
     with pytest.raises(RegistryObjectNotFound):
-        registry.delete("test:nonexistent", version=None)
+        registry.delete("test:nonexistent")
 
 
 # ─────────────────────────────────────────────────────────────────────────────