fix: dedupe and report automatic template creation from images

install_appliances_from_image relied on the name+version pair check in
TemplatesService, so the same appliance reached through a second image
(the CSR1000v case) created a second template sharing the name. The auto
path now skips when any template with the same name exists, whatever the
version, and returns a manifest of created and skipped candidates;
POST /images/install replies 200 with that manifest instead of an empty
204, and the image_install MCP tool surfaces it.
This commit is contained in:
YueGuobin committed 2026-08-26 00:04:57 +08:00
1 parent 888afdccbd
commit 8a8314ab29
7 files changed
+154 -11

No files matched your search

+3 -1
View File
@@ -1435,7 +1435,9 @@ async def image_install() -> list[dict[str, Any]]:
This is NOT for downloading images. Images must be uploaded first (via the GNS3 Web UI).
If an uploaded image matches a known appliance, a template is automatically created.
Images already referenced by existing templates are skipped.
Returns {"created": [...], "skipped": [...]}: images already referenced by existing
templates are skipped, and no template is auto-created when one with the same name
already exists (regardless of version).
"""
return await asyncio.to_thread(_run_handler_sync, install_images_handler, {})
+5 -2
View File
@@ -72,6 +72,9 @@ def prune_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> di
def install_images_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]:
conn = _get_connector(gns3_ctx)
# Returns 204 No Content on success (empty body, no .json())
conn.http_call("post", f"{conn.base_url}/images/install")
response = conn.http_call("post", f"{conn.base_url}/images/install")
if response.content:
# the install endpoint reports which templates were created or skipped
return response.json()
# tolerate an empty body in case an older server still replies with 204
return {"message": "Image installation completed"}
+18 -3
View File
@@ -206,19 +206,24 @@ async def prune_images(
@router.post(
"/install",
status_code=status.HTTP_204_NO_CONTENT,
status_code=status.HTTP_200_OK,
dependencies=[Depends(has_privilege("Image.Allocate"))]
)
async def install_images(
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository))
) -> None:
) -> dict:
"""
Attempt to automatically create templates based on image checksums.
Returns the list of created templates and the list of skipped
candidates (with the reason why they were skipped).
Required privilege: Image.Allocate
"""
created = []
skipped = []
skip_images = get_builtin_disks()
images = await images_repo.get_images()
for image in images:
@@ -229,8 +234,12 @@ async def install_images(
if templates:
# the image is already used by a template
log.warning(f"Image '{image.path}' is used by one or more templates")
skipped.append({
"name": image.filename,
"reason": "image is already used by one or more templates",
})
continue
await Controller.instance().appliance_manager.install_appliances_from_image(
results = await Controller.instance().appliance_manager.install_appliances_from_image(
image.path,
image.checksum,
images_repo,
@@ -239,6 +248,12 @@ async def install_images(
None,
os.path.dirname(image.path)
)
for result in results:
if result.get("status") == "created":
created.append({k: v for k, v in result.items() if k != "status"})
else:
skipped.append({k: v for k, v in result.items() if k != "status"})
return {"created": created, "skipped": skipped}
@router.get(
+41 -3
View File
@@ -245,11 +245,16 @@ class ApplianceManager:
rbac_repo: RbacRepository,
current_user: schemas.User,
image_dir: str
) -> None:
) -> List[dict]:
"""
Install appliances using an image checksum
Install appliances using an image checksum.
Returns a manifest of what happened: one entry per attempted template,
either {"status": "created", ...template fields} or
{"status": "skipped", "name", "reason"}.
"""
results: List[dict] = []
appliances_info = self._find_appliances_from_image_checksum(image_checksum)
for appliance, image_version in appliances_info:
try:
@@ -257,15 +262,48 @@ class ApplianceManager:
ApplianceModel.model_validate(appliance.asdict())
except ValidationError as e:
log.warning(f"Could not validate appliance '{appliance.id}': {e}")
results.append({
"status": "skipped",
"name": appliance.name,
"reason": f"could not validate appliance '{appliance.id}': {e}",
})
continue
if appliance.versions:
for version in appliance.versions:
if version.get("name") == image_version:
try:
await self._find_appliance_version_images(appliance, version, images_repo, image_dir)
template_data = await self._appliance_to_template(appliance, version)
await self._create_template(template_data, templates_repo, rbac_repo, current_user)
name = template_data.get("name")
existing = await templates_repo.get_template_by_name(name) if name else None
if existing is not None:
# never automatically create a second template with the same
# name: the name+version check in TemplatesService would allow
# duplicates when the appliance version differs, but two
# templates sharing a name is never what the user asked for here
log.warning(f"Template '{name}' already exists, skipping automatic template creation")
results.append({
"status": "skipped",
"name": name,
"reason": f"a template named '{name}' already exists",
})
continue
template = await self._create_template(template_data, templates_repo, rbac_repo, current_user)
results.append({
"status": "created",
"template_id": str(template.get("template_id")),
"name": template.get("name"),
"version": template.get("version"),
"template_type": template.get("template_type"),
})
except (ControllerError, InvalidImageError) as e:
log.warning(f"Could not automatically create template using image '{image_path}': {e}")
results.append({
"status": "skipped",
"name": appliance.name,
"reason": str(e),
})
return results
async def install_appliance(
self,
+11
View File
@@ -71,6 +71,17 @@ class TemplatesRepository(BaseRepository):
result = await self._db_session.execute(query)
return result.scalars().first()
async def get_template_by_name(self, name: str) -> Union[None, models.Template]:
"""
Return the first template with this name, regardless of version.
"""
query = select(models.Template).\
options(selectinload(models.Template.images)).\
where(models.Template.name == name)
result = await self._db_session.execute(query)
return result.scalars().first()
async def get_templates(self) -> List[models.Template]:
query = select(models.Template).options(selectinload(models.Template.images))