mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-11-03 05:51:15 +02:00
feat(api): add exact-match name filter to list endpoints
This commit is contained in:
11 files changed
+167
-6
No files matched your search
@@ -21,7 +21,7 @@ API routes for computes.
|
||||
from typing import Any, List, Optional, Union
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, status
|
||||
from fastapi import APIRouter, Body, Depends, Query, status
|
||||
|
||||
import gns3server.db.models as models
|
||||
from gns3server import schemas
|
||||
@@ -106,14 +106,21 @@ async def get_compute(
|
||||
)
|
||||
async def get_computes(
|
||||
computes_repo: ComputesRepository = Depends(get_repository(ComputesRepository)),
|
||||
name: Optional[str] = Query(None, description="Return only computes whose name exactly matches (case-sensitive)"),
|
||||
) -> List[models.Compute]:
|
||||
"""
|
||||
Return all computes known by the controller.
|
||||
|
||||
Required privilege: Compute.Audit
|
||||
|
||||
Query Parameters:
|
||||
- name: Exact, case-sensitive match on the compute name.
|
||||
"""
|
||||
|
||||
return await ComputesService(computes_repo).get_computes()
|
||||
computes = await ComputesService(computes_repo).get_computes()
|
||||
if name is not None:
|
||||
computes = [compute for compute in computes if compute.name == name]
|
||||
return computes
|
||||
|
||||
|
||||
@router.put(
|
||||
|
||||
@@ -195,14 +195,18 @@ async def get_images(
|
||||
images_repo: ImagesRepository = Depends(get_repository(ImagesRepository)),
|
||||
image_type: Optional[schemas.ImageType] = None,
|
||||
availability: Optional[Literal["unknown", "available", "missing", "unavailable", "invalid"]] = None,
|
||||
name: Optional[str] = Query(None, description="Return only images whose filename exactly matches (case-sensitive)"),
|
||||
) -> List[models.Image]:
|
||||
"""
|
||||
Return all images.
|
||||
|
||||
Required privilege: Image.Audit
|
||||
|
||||
Query Parameters:
|
||||
- name: Exact, case-sensitive match on the image filename.
|
||||
"""
|
||||
|
||||
return await images_repo.get_images(image_type, availability)
|
||||
return await images_repo.get_images(image_type, availability, name)
|
||||
|
||||
|
||||
@router.post(
|
||||
|
||||
@@ -151,6 +151,7 @@ async def create_node(node_create: schemas.NodeCreate, project: Project = Depend
|
||||
def get_nodes(
|
||||
project: Project = Depends(dep_project),
|
||||
tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
|
||||
name: Optional[str] = Query(None, description="Return only nodes whose name exactly matches (case-sensitive)"),
|
||||
) -> List[schemas.Node]:
|
||||
"""
|
||||
Return all nodes belonging to a given project.
|
||||
@@ -160,6 +161,7 @@ def get_nodes(
|
||||
Query Parameters:
|
||||
- tags: Filter by tags. Multiple tags are ANDed together.
|
||||
Example: ?tags=vendor:cisco&tags=model:7200
|
||||
- name: Exact, case-sensitive match on the node name. Combined with other filters using AND.
|
||||
"""
|
||||
|
||||
if project.status == "closed":
|
||||
@@ -168,6 +170,9 @@ def get_nodes(
|
||||
else:
|
||||
nodes = [v.asdict() for v in project.nodes.values()]
|
||||
|
||||
if name is not None:
|
||||
nodes = [node for node in nodes if node.get("name") == name]
|
||||
|
||||
# Filter by tags if provided (all filter tags have to match the node tags)
|
||||
if tags:
|
||||
filtered_nodes = []
|
||||
|
||||
@@ -34,7 +34,7 @@ log = logging.getLogger()
|
||||
from typing import Any, List, Optional
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect, status
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, WebSocket, WebSocketDisconnect, status
|
||||
from fastapi.encoders import jsonable_encoder
|
||||
from fastapi.responses import FileResponse, StreamingResponse
|
||||
from websockets.exceptions import ConnectionClosed, WebSocketException
|
||||
@@ -78,11 +78,15 @@ def dep_project(project_id: UUID) -> Project:
|
||||
async def get_projects(
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
|
||||
name: Optional[str] = Query(None, description="Return only projects whose name exactly matches (case-sensitive)"),
|
||||
) -> List[schemas.Project]:
|
||||
"""
|
||||
Return all projects.
|
||||
|
||||
Required privilege: Project.Audit
|
||||
|
||||
Query Parameters:
|
||||
- name: Exact, case-sensitive match on the project name.
|
||||
"""
|
||||
|
||||
controller = Controller.instance()
|
||||
@@ -91,7 +95,7 @@ async def get_projects(
|
||||
|
||||
if current_user.is_superadmin:
|
||||
# super admin sees all projects
|
||||
return [p.asdict() for p in controller.projects.values()]
|
||||
return [p.asdict() for p in controller.projects.values() if name is None or p.name == name]
|
||||
|
||||
# Batch ACE + resource pool check (3 DB queries regardless of project count)
|
||||
all_project_ids = list(controller.projects.keys())
|
||||
@@ -102,6 +106,8 @@ async def get_projects(
|
||||
# Step 2: Filter direct ACE projects by created_by
|
||||
# Direct project sharing is only available through resource pools
|
||||
for p in controller.projects.values():
|
||||
if name is not None and p.name != name:
|
||||
continue
|
||||
if p.id in direct_ace_ids and p.created_by == current_user.username:
|
||||
if p.id not in seen_project_ids:
|
||||
projects.append(p.asdict())
|
||||
@@ -109,6 +115,8 @@ async def get_projects(
|
||||
|
||||
# Step 3: Resource pool projects (no created_by filter)
|
||||
for p in controller.projects.values():
|
||||
if name is not None and p.name != name:
|
||||
continue
|
||||
if p.id in pool_accessible_ids:
|
||||
if p.id not in seen_project_ids:
|
||||
projects.append(p.asdict())
|
||||
|
||||
@@ -195,6 +195,7 @@ async def get_templates(
|
||||
templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
|
||||
current_user: schemas.User = Depends(get_current_active_user),
|
||||
tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)"),
|
||||
name: Optional[str] = Query(None, description="Return only templates whose name exactly matches (case-sensitive)"),
|
||||
) -> List[dict]:
|
||||
"""
|
||||
Return all templates.
|
||||
@@ -204,10 +205,14 @@ async def get_templates(
|
||||
Query Parameters:
|
||||
- tags: Filter by tags. Multiple tags are ANDed together.
|
||||
Example: ?tags=vendor:cisco&tags=model:7200
|
||||
- name: Exact, case-sensitive match on the template name. Combined with other filters using AND.
|
||||
"""
|
||||
|
||||
templates = await TemplatesService(templates_repo).get_templates()
|
||||
|
||||
if name is not None:
|
||||
templates = [template for template in templates if template.get("name") == name]
|
||||
|
||||
# Filter by tags if provided (all filter tags have to match the node tags)
|
||||
if tags:
|
||||
filtered_templates = []
|
||||
|
||||
Reference in new issue
Block a user