Allow project admins to delete imports

docs/docs/reference/cli/dstack/import.md

@@ -17,3 +17,18 @@ $ dstack import list --help
 ```
 
 </div>
+
+## dstack import delete
+
+The `dstack import delete` command deletes the specified import. This makes the imported resources unavailable in your project, while they still exist in the host project.
+
+##### Usage
+
+<div class="termy">
+
+```shell
+$ dstack import delete --help
+#GENERATE#
+```
+
+</div>

src/dstack/_internal/cli/commands/import_.py

@@ -3,7 +3,8 @@
 from rich.table import Table
 
 from dstack._internal.cli.commands import APIBaseCommand
-from dstack._internal.cli.utils.common import add_row_from_dict, console
+from dstack._internal.cli.services.completion import ImportNameCompleter
+from dstack._internal.cli.utils.common import add_row_from_dict, confirm_ask, console
 from dstack._internal.core.models.imports import Import
 
 
@@ -21,6 +22,18 @@ def _register(self):
         )
         list_parser.set_defaults(subfunc=self._list)
 
+        delete_parser = subparsers.add_parser(
+            "delete", help="Delete an import", formatter_class=self._parser.formatter_class
+        )
+        delete_parser.add_argument(
+            "name",
+            help="The import to delete, in `export-project/export-name` format",
+        ).completer = ImportNameCompleter()  # type: ignore[attr-defined]
+        delete_parser.add_argument(
+            "-y", "--yes", help="Don't ask for confirmation", action="store_true"
+        )
+        delete_parser.set_defaults(subfunc=self._delete)
+
     def _command(self, args: argparse.Namespace):
         super()._command(args)
         args.subfunc(args)
@@ -29,6 +42,27 @@ def _list(self, args: argparse.Namespace):
         imports = self.api.client.imports.list(self.api.project)
         print_imports_table(imports)
 
+    def _delete(self, args: argparse.Namespace):
+        parts = args.name.split("/")
+        if len(parts) != 2 or not parts[0] or not parts[1]:
+            self._parser.error(
+                f"Invalid format: {args.name!r}. Expected <export-project>/<export-name>"
+            )
+        export_project_name, export_name = parts
+
+        if not args.yes and not confirm_ask(f"Delete the import [code]{args.name}[/]?"):
+            console.print("\nExiting...")
+            return
+
+        with console.status("Deleting import..."):
+            self.api.client.imports.delete(
+                project_name=self.api.project,
+                export_project_name=export_project_name,
+                export_name=export_name,
+            )
+
+        console.print(f"Import [code]{args.name}[/] deleted")
+
 
 def print_imports_table(imports: list[Import]):
     table = Table(box=None)

src/dstack/_internal/cli/services/completion.py

@@ -85,6 +85,14 @@ def fetch_resource_names(self, api: Client) -> Iterable[str]:
         return [r.name for r in api.client.exports.list(api.project)]
 
 
+class ImportNameCompleter(BaseAPINameCompleter):
+    def fetch_resource_names(self, api: Client) -> Iterable[str]:
+        return [
+            f"{imp.export.project_name}/{imp.export.name}"
+            for imp in api.client.imports.list(api.project)
+        ]
+
+
 class ProjectNameCompleter(BaseCompleter):
     """
     Completer for local project names.

src/dstack/_internal/server/routers/imports.py

@@ -6,7 +6,8 @@
 from dstack._internal.core.models.imports import Import
 from dstack._internal.server.db import get_session
 from dstack._internal.server.models import ProjectModel, UserModel
-from dstack._internal.server.security.permissions import ProjectMember
+from dstack._internal.server.schemas.imports import DeleteImportRequest
+from dstack._internal.server.security.permissions import ProjectAdmin, ProjectMember
 from dstack._internal.server.services import imports as imports_services
 from dstack._internal.server.utils.routers import get_base_api_additional_responses
 
@@ -17,6 +18,21 @@
 )
 
 
+@project_router.post("/delete")
+async def delete_import(
+    body: DeleteImportRequest,
+    session: Annotated[AsyncSession, Depends(get_session)],
+    user_project: Annotated[tuple[UserModel, ProjectModel], Depends(ProjectAdmin())],
+):
+    _, project = user_project
+    await imports_services.delete_import(
+        session=session,
+        project=project,
+        export_name=body.export_name,
+        export_project_name=body.export_project_name,
+    )
+
+
 @project_router.post("/list", response_model=list[Import])
 async def list_imports(
     session: Annotated[AsyncSession, Depends(get_session)],

src/dstack/_internal/server/schemas/imports.py

@@ -0,0 +1,10 @@
+from dstack._internal.core.models.common import CoreModel
+
+
+class DeleteImportRequest(CoreModel):
+    """
+    Imports are unnamed, so they are deleted using the name and project of their export.
+    """
+
+    export_name: str
+    export_project_name: str

src/dstack/_internal/server/services/imports.py

@@ -2,6 +2,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload, selectinload
 
+from dstack._internal.core.errors import ResourceNotExistsError
 from dstack._internal.core.models.imports import Import, ImportExport, ImportExportedFleet
 from dstack._internal.server.models import (
     ExportedFleetModel,
@@ -10,6 +11,8 @@
     ImportModel,
     ProjectModel,
 )
+from dstack._internal.server.services.exports import get_export_model_by_name_for_update
+from dstack._internal.server.services.projects import get_project_model_by_name
 
 
 async def list_imports(session: AsyncSession, project: ProjectModel) -> list[Import]:
@@ -36,6 +39,33 @@ async def list_imports(session: AsyncSession, project: ProjectModel) -> list[Imp
     return [import_model_to_import(imp) for imp in imports]
 
 
+async def delete_import(
+    session: AsyncSession,
+    project: ProjectModel,
+    export_name: str,
+    export_project_name: str,
+) -> None:
+    # Always the same error, so as not to expose the existence of exports
+    # that are not imported in this project.
+    not_found_error = ResourceNotExistsError(
+        f"Import '{export_project_name}/{export_name}' not found in project {project.name!r}"
+    )
+    exporter_project = await get_project_model_by_name(session, export_project_name)
+    if exporter_project is None:
+        raise not_found_error
+    async with get_export_model_by_name_for_update(
+        session, exporter_project, export_name
+    ) as export:
+        if export is None:
+            raise not_found_error
+        if project.name.lower() not in {imp.project.name.lower() for imp in export.imports}:
+            raise not_found_error
+        export.imports = [
+            imp for imp in export.imports if imp.project.name.lower() != project.name.lower()
+        ]
+        await session.commit()
+
+
 def import_model_to_import(import_model: ImportModel) -> Import:
     return Import(
         id=import_model.id,

src/dstack/api/server/_imports.py

@@ -3,10 +3,17 @@
 from pydantic import parse_obj_as
 
 from dstack._internal.core.models.imports import Import
+from dstack._internal.server.schemas.imports import DeleteImportRequest
 from dstack.api.server._group import APIClientGroup
 
 
 class ImportsAPIClient(APIClientGroup):
     def list(self, project_name: str) -> List[Import]:
         resp = self._request(f"/api/project/{project_name}/imports/list")
         return parse_obj_as(List[Import.__response__], resp.json())
+
+    def delete(self, *, project_name: str, export_project_name: str, export_name: str) -> None:
+        body = DeleteImportRequest(
+            export_project_name=export_project_name, export_name=export_name
+        )
+        self._request(f"/api/project/{project_name}/imports/delete", body=body.json())

src/tests/_internal/server/routers/test_imports.py

@@ -2,9 +2,11 @@
 
 import pytest
 from httpx import AsyncClient
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from dstack._internal.core.models.users import GlobalRole, ProjectRole
+from dstack._internal.server.models import ExportModel, ImportModel
 from dstack._internal.server.services.projects import add_project_member
 from dstack._internal.server.testing.common import (
     create_export,
@@ -23,6 +25,119 @@
 ]
 
 
+class TestDeleteImport:
+    async def test_returns_403_if_not_authenticated(self, client: AsyncClient):
+        response = await client.post(
+            "/api/project/TestProject/imports/delete",
+            json={"export_name": "test-export", "export_project_name": "ExporterProject"},
+        )
+        assert response.status_code in [401, 403]
+
+    async def test_returns_403_if_not_admin(self, session: AsyncSession, client: AsyncClient):
+        user = await create_user(session=session, global_role=GlobalRole.USER)
+        exporter_project = await create_project(
+            session=session, name="ExporterProject", owner=user
+        )
+        importer_project = await create_project(
+            session=session, name="ImporterProject", owner=user
+        )
+        # The user is admin of the exporter project, but not of the importer
+        await add_project_member(
+            session=session, project=exporter_project, user=user, project_role=ProjectRole.ADMIN
+        )
+        await add_project_member(
+            session=session, project=importer_project, user=user, project_role=ProjectRole.USER
+        )
+        response = await client.post(
+            f"/api/project/{importer_project.name}/imports/delete",
+            headers=get_auth_headers(user.token),
+            json={"export_name": "test-export", "export_project_name": "ExporterProject"},
+        )
+        assert response.status_code == 403
+
+    async def test_deletes_import(self, session: AsyncSession, client: AsyncClient):
+        user = await create_user(session=session, global_role=GlobalRole.USER)
+        importer_project = await create_project(
+            session=session, name="ImporterProject", owner=user
+        )
+        await add_project_member(
+            session=session, project=importer_project, user=user, project_role=ProjectRole.ADMIN
+        )
+        exporter_project = await create_project(session=session, name="ExporterProject")
+        fleet = await create_fleet(
+            session=session,
+            project=exporter_project,
+            name="fleet1",
+            spec=get_fleet_spec(get_ssh_fleet_configuration()),
+        )
+        await create_export(
+            session=session,
+            exporter_project=exporter_project,
+            importer_projects=[importer_project],
+            exported_fleets=[fleet],
+            name="test-export",
+        )
+
+        response = await client.post(
+            f"/api/project/{importer_project.name}/imports/delete",
+            headers=get_auth_headers(user.token),
+            json={
+                "export_name": "test-export",
+                "export_project_name": "ExPoRtErPrOjEcT",  # case-insensitive
+            },
+        )
+        assert response.status_code == 200
+
+        res = await session.execute(select(func.count()).select_from(ImportModel))
+        assert res.scalar_one() == 0
+        res = await session.execute(select(func.count()).select_from(ExportModel))
+        assert res.scalar_one() == 1
+
+    async def test_returns_400_for_nonexistent_import(
+        self, session: AsyncSession, client: AsyncClient
+    ):
+        user = await create_user(session=session, global_role=GlobalRole.USER)
+        importer_project = await create_project(
+            session=session, name="ImporterProject", owner=user
+        )
+        await add_project_member(
+            session=session, project=importer_project, user=user, project_role=ProjectRole.ADMIN
+        )
+
+        exporter_project = await create_project(
+            session=session, name="ExporterProject", owner=user
+        )
+        await create_export(
+            session=session,
+            exporter_project=exporter_project,
+            importer_projects=[],
+            exported_fleets=[],
+            name="test-export",
+        )
+
+        async def assert_not_found(export_project_name, export_name):
+            response = await client.post(
+                f"/api/project/{importer_project.name}/imports/delete",
+                headers=get_auth_headers(user.token),
+                json={"export_name": export_name, "export_project_name": export_project_name},
+            )
+            assert response.status_code == 400
+            assert response.json()["detail"][0]["code"] == "resource_not_exists"
+            # The error should be the same regardless of what wasn't found
+            # (the exporter, the export, or the import),
+            # so that users cannot infer the existence of exports they are not given access to.
+            assert response.json()["detail"][0]["msg"] == (
+                f"Import '{export_project_name}/{export_name}' not found in project 'ImporterProject'"
+            )
+
+        # Exporter not found
+        await assert_not_found(export_project_name="WrongProject", export_name="test-export")
+        # Export not found
+        await assert_not_found(export_project_name="ExporterProject", export_name="wrong-export")
+        # Import not found
+        await assert_not_found(export_project_name="ExporterProject", export_name="test-export")
+
+
 class TestListImports:
     async def test_returns_403_if_not_authenticated(self, client: AsyncClient):
         response = await client.post(