Core: Services: Kraken: Allow fetching availbale versions of a given Extension

core/services/kraken/api/v2/routers/manifest.py

@@ -76,6 +76,30 @@ async def fetch_consolidated() -> list[RepositoryEntry]:
     return await manifest_manager.fetch_consolidated()
 
 
+@manifest_router_v2.get("/tags/{manifest_identifier}/{extension_identifier}/", status_code=status.HTTP_200_OK)
+@manifest_to_http_exception
+async def fetch_ext_tags_from_manifest(
+    manifest_identifier: str, extension_identifier: str, stable: bool = False
+) -> list[str]:
+    """
+    Get all tags for a given extension identifier using one manifest as source.
+    """
+    versions = await manifest_manager.fetch_extension_versions(extension_identifier, stable, manifest_identifier)
+
+    return [str(version) for version in versions]
+
+
+@manifest_router_v2.get("/tags/{extension_identifier}", status_code=status.HTTP_200_OK)
+@manifest_to_http_exception
+async def fetch_ext_tags_from_consolidated(extension_identifier: str, stable: bool = False) -> list[str]:
+    """
+    Get all tags for a given extension identifier using consolidated manifest as source.
+    """
+    versions = await manifest_manager.fetch_extension_versions(extension_identifier, stable)
+
+    return [str(version) for version in versions]
+
+
 @manifest_router_v2.post("/", status_code=status.HTTP_201_CREATED)
 @manifest_to_http_exception
 async def create(body: ManifestSource, validate_url: bool = True) -> Manifest:

core/services/kraken/manifest/manifest.py

@@ -247,17 +247,22 @@ async def order_sources(self, identifiers: List[str]) -> None:
         self._settings.manifests = ordered_manifests
         self._manager.save()
 
-    async def fetch_extension(self, identifier: str) -> Optional[RepositoryEntry]:
-        # Only fetch enabled sources already sorted by priority
-        manifest = await self.fetch_consolidated()
-        return next((ext for ext in manifest if ext.identifier == identifier), None)
-
-    async def fetch_latest_extension_version(self, identifier: str, stable: bool) -> Optional[ExtensionVersion]:
-        manifest = await self.fetch_consolidated()
-        ext = next((ext for ext in manifest if ext.identifier == identifier), None)
-
+    async def fetch_extension(self, extension_id: str, manifest_id: Optional[str] = None) -> Optional[RepositoryEntry]:
+        manifest = []
+        if manifest_id is None:
+            # Only fetch enabled sources already sorted by priority
+            manifest = await self.fetch_consolidated()
+        else:
+            manifest = (await self.fetch_by_identifier(manifest_id, fetch_data=True)).data or []
+
+        return next((ext for ext in manifest if ext.identifier == extension_id), None)
+
+    async def fetch_extension_versions(
+        self, extension_id: str, stable: bool, manifest_id: Optional[str] = None
+    ) -> List[semver.VersionInfo]:
+        ext = await self.fetch_extension(extension_id, manifest_id)
         if not ext or not ext.versions:
-            return None
+            return []
 
         def valid_semver(string: str) -> Optional[semver.VersionInfo]:
             # We want to allow versions to be prefixed with a 'v'.
@@ -266,17 +271,27 @@ def valid_semver(string: str) -> Optional[semver.VersionInfo]:
             try:
                 return semver.VersionInfo.parse(string)
             except ValueError:
-                return None
+                return []
 
         versions: List[semver.VersionInfo] = sorted([valid_semver(tag) for tag in ext.versions], reverse=True)
-
         if stable:
             versions = [v for v in versions if not v.prerelease and not v.patch]
 
+        return versions
+
+    async def fetch_latest_extension_version(
+        self, extension_id: str, stable: bool, manifest_id: Optional[str] = None
+    ) -> Optional[ExtensionVersion]:
+        ext = await self.fetch_extension(extension_id, manifest_id)
+        if not ext or not ext.versions:
+            return None
+
+        versions = await self.fetch_extension_versions(extension_id, stable, manifest_id)
+
         return ext.versions.get(str(versions[0])) if versions else None
 
-    async def fetch_extension_version(self, identifier: str, tag: str) -> Optional[ExtensionVersion]:
-        ext = await self.fetch_extension(identifier)
+    async def fetch_extension_version(self, extension_id: str, tag: str) -> Optional[ExtensionVersion]:
+        ext = await self.fetch_extension(extension_id)
         if not ext:
             return None