Use lists for tags instead of dicts

This commit is contained in:
grossmj 2026-02-23 23:26:42 +08:00
parent 66f32cecf2
commit bdcb6445c7
No known key found for this signature in database
GPG Key ID: 1E7DD6DBB53FF3D7
6 changed files with 28 additions and 49 deletions

View File

@ -137,7 +137,7 @@ async def create_node(node_data: schemas.NodeCreate, project: Project = Depends(
) )
def get_nodes( def get_nodes(
project: Project = Depends(dep_project), project: Project = Depends(dep_project),
tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g., tags=vendor:cisco&tags=model:7200)") tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)")
) -> List[schemas.Node]: ) -> List[schemas.Node]:
""" """
Return all nodes belonging to a given project. Return all nodes belonging to a given project.
@ -145,7 +145,7 @@ def get_nodes(
Required privilege: Node.Audit Required privilege: Node.Audit
Query Parameters: Query Parameters:
- tags: Filter by tags in format "key:value". Multiple tags are ANDed together. - tags: Filter by tags. Multiple tags are ANDed together.
Example: ?tags=vendor:cisco&tags=model:7200 Example: ?tags=vendor:cisco&tags=model:7200
""" """
@ -155,29 +155,19 @@ def get_nodes(
else: else:
nodes = [v.asdict() for v in project.nodes.values()] nodes = [v.asdict() for v in project.nodes.values()]
# Filter by tags if provided # Filter by tags if provided (all filter tags have to match the node tags)
if tags: if tags:
filtered_nodes = [] filtered_nodes = []
for node in nodes: for node in nodes:
node_dict = node.asdict() if hasattr(node, 'asdict') else node node_tags = node.get("tags") or []
node_tags = node_dict.get("tags") or {}
# Check if all tag filters match
match = True match = True
for tag_filter in tags: for tag_filter in tags:
if ":" in tag_filter: if tag_filter not in node_tags:
key, value = tag_filter.split(":", 1) match = False
if node_tags.get(key) != value: break
match = False
break
else:
# Check if key exists
if tag_filter not in node_tags:
match = False
break
if match: if match:
filtered_nodes.append(node) filtered_nodes.append(node)
nodes = filtered_nodes return filtered_nodes
return nodes return nodes

View File

@ -170,7 +170,7 @@ async def get_templates(
templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)), templates_repo: TemplatesRepository = Depends(get_repository(TemplatesRepository)),
current_user: schemas.User = Depends(get_current_active_user), current_user: schemas.User = Depends(get_current_active_user),
rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)), rbac_repo: RbacRepository = Depends(get_repository(RbacRepository)),
tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g., tags=vendor:cisco&tags=model:7200)") tags: Optional[List[str]] = Query(None, description="Filter by tags (e.g. tags=vendor:cisco&tags=model:7200)")
) -> List[schemas.Template]: ) -> List[schemas.Template]:
""" """
Return all templates. Return all templates.
@ -178,30 +178,22 @@ async def get_templates(
Required privilege: Template.Audit Required privilege: Template.Audit
Query Parameters: Query Parameters:
- tags: Filter by tags in format "key:value". Multiple tags are ANDed together. - tags: Filter by tags. Multiple tags are ANDed together.
Example: ?tags=vendor:cisco&tags=model:7200 Example: ?tags=vendor:cisco&tags=model:7200
""" """
templates = await TemplatesService(templates_repo).get_templates() templates = await TemplatesService(templates_repo).get_templates()
# Filter by tags if provided # Filter by tags if provided (all filter tags have to match the node tags)
if tags: if tags:
filtered_templates = [] filtered_templates = []
for template in templates: for template in templates:
template_tags = template.get("tags") or {} template_tags = template.get("tags") or []
# Check if all tag filters match
match = True match = True
for tag_filter in tags: for tag_filter in tags:
if ":" in tag_filter: if tag_filter not in template_tags:
key, value = tag_filter.split(":", 1) match = False
if template_tags.get(key) != value: break
match = False
break
else:
# Check if key exists
if tag_filter not in template_tags:
match = False
break
if match: if match:
filtered_templates.append(template) filtered_templates.append(template)
templates = filtered_templates templates = filtered_templates

View File

@ -97,7 +97,7 @@ class Node:
self._y = 0 self._y = 0
self._z = 1 # default z value is 1 self._z = 1 # default z value is 1
self._locked = False self._locked = False
self._tags = {} self._tags = []
self._ports = None self._ports = None
self._symbol = None self._symbol = None
self._custom_adapters = [] self._custom_adapters = []
@ -225,10 +225,7 @@ class Node:
@tags.setter @tags.setter
def tags(self, val): def tags(self, val):
if isinstance(val, dict): self._tags = val
self._tags = val
else:
self._tags = {}
def _base_config_file_content(self, path): def _base_config_file_content(self, path):
if not os.path.isabs(path): if not os.path.isabs(path):

View File

@ -36,7 +36,7 @@ class Template(BaseTable):
builtin = Column(Boolean, default=False) builtin = Column(Boolean, default=False)
usage = Column(String) usage = Column(String)
template_type = Column(String) template_type = Column(String)
tags = Column(JSON, default='{}') tags = Column(JSON)
compute_id = Column(String) compute_id = Column(String)
images = relationship("Image", secondary=image_template_map, back_populates="templates") images = relationship("Image", secondary=image_template_map, back_populates="templates")

View File

@ -14,8 +14,8 @@
# You should have received a copy of the GNU General Public License # You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
from pydantic import BaseModel, Field, model_validator from pydantic import BaseModel, Field
from typing import List, Optional, Union, Any, Dict from typing import List, Optional, Union
from enum import Enum from enum import Enum
from uuid import UUID, uuid4 from uuid import UUID, uuid4
@ -128,14 +128,14 @@ class NodeBase(BaseModel):
z: Optional[int] = 1 z: Optional[int] = 1
locked: Optional[bool] = Field(False, description="Whether the element locked or not") locked: Optional[bool] = Field(False, description="Whether the element locked or not")
port_name_format: Optional[str] = Field( port_name_format: Optional[str] = Field(
None, descript_port_name_formation="Formatting for port name {0} will be replace by port number" None, description="Formatting for port name {0} will be replace by port number"
) )
port_segment_size: Optional[int] = Field(None, description="Size of the port segment") port_segment_size: Optional[int] = Field(None, description="Size of the port segment")
first_port_name: Optional[str] = Field(None, description="Name of the first port") first_port_name: Optional[str] = Field(None, description="Name of the first port")
custom_adapters: Optional[List[CustomAdapter]] = None custom_adapters: Optional[List[CustomAdapter]] = None
tags: Optional[Dict[str, str]] = Field( tags: Optional[List[str]] = Field(
default_factory=dict, default_factory=list,
description="User-defined metadata tags inherited from template or custom" description="User-defined metadata tags (e.g. 'vendor:cisco' or 'model:7200')"
) )

View File

@ -15,7 +15,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
from pydantic import ConfigDict, BaseModel, Field from pydantic import ConfigDict, BaseModel, Field
from typing import Optional, Union, Dict from typing import Optional, List
from enum import Enum from enum import Enum
from uuid import UUID from uuid import UUID
@ -48,9 +48,9 @@ class TemplateBase(BaseModel):
template_type: Optional[NodeType] = None template_type: Optional[NodeType] = None
compute_id: Optional[str] = None compute_id: Optional[str] = None
usage: Optional[str] = "" usage: Optional[str] = ""
tags: Optional[Dict[str, str]] = Field( tags: Optional[List[str]] = Field(
default_factory=dict, default_factory=list,
description="User-defined metadata tags as key-value pairs (e.g., {'vendor': 'cisco', 'model': '7200', 'netmiko_device_type': 'cisco_ios'})" description="User-defined metadata tags (e.g. 'vendor:cisco' or 'model:7200')"
) )