Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 36 additions & 3 deletions nsc/cli/handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,33 @@ def _out(stream: TextIO | None) -> TextIO:
return stream if stream is not None else sys.stdout


def _custom_field_header_labels(
ctx: RuntimeContext, operation: Operation, columns: list[str] | None
) -> dict[str, str] | None:
"""Map ``custom_fields.<name>`` columns to their NetBox labels, or None.

Only fetches definitions when a custom-field column is actually selected, so
ordinary lists incur no extra request. Display-only: the raw key stays the
selector everywhere else.
"""
if not columns or not any(c.startswith("custom_fields.") for c in columns):
return None
from nsc.savedfilters.custom_fields import ( # noqa: PLC0415
CustomFieldResolver,
custom_field_labels,
)

# Custom-field definitions are keyed by the collection (list) endpoint, so a
# detail path like /api/dcim/devices/{id}/ must reduce to /api/dcim/devices/.
list_path = operation.path
if list_path.endswith("{id}/"):
list_path = list_path[: -len("{id}/")]
defs = CustomFieldResolver(ctx.client).resolve(list_path)
if defs is None:
return None
return custom_field_labels(columns, defs)


def parse_filters(raw: list[str]) -> dict[str, str]:
out: dict[str, str] = {}
for item in raw:
Expand Down Expand Up @@ -97,10 +124,12 @@ def handle_list(
page_size=ctx.page_size,
)
)
columns = ctx.resolve_columns(op_tag, op_resource, operation)
render(
rows,
format=ctx.output_format,
columns=ctx.resolve_columns(op_tag, op_resource, operation),
columns=columns,
header_labels=_custom_field_header_labels(ctx, operation, columns),
stream=_out(stream),
compact=ctx.compact,
color=ctx.color,
Expand All @@ -124,10 +153,12 @@ def handle_get(
try:
params, path_vars = _split_params(operation, kwargs)
obj = ctx.client.get(operation.path.format(**path_vars), params)
columns = ctx.resolve_columns(op_tag, op_resource, operation)
render(
obj,
format=ctx.output_format,
columns=ctx.resolve_columns(op_tag, op_resource, operation),
columns=columns,
header_labels=_custom_field_header_labels(ctx, operation, columns),
stream=_out(stream),
compact=ctx.compact,
color=ctx.color,
Expand Down Expand Up @@ -611,10 +642,12 @@ def _render_response(
if is_delete:
_render_delete_ok(ctx, stream=stream)
return
columns = ctx.resolve_columns(op_tag, op_resource, operation)
render(
response,
format=ctx.output_format,
columns=ctx.resolve_columns(op_tag, op_resource, operation),
columns=columns,
header_labels=_custom_field_header_labels(ctx, operation, columns),
stream=stream,
compact=ctx.compact,
color=ctx.color,
Expand Down
8 changes: 7 additions & 1 deletion nsc/output/csv_.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,14 +14,20 @@ def render(
*,
stream: TextIO = sys.stdout,
columns: list[str] | None = None,
header_labels: dict[str, str] | None = None,
) -> None:
records = [data] if isinstance(data, dict) else list(data)
if not records:
return
flat_records = [flatten(r, columns=columns) for r in records]
fieldnames = columns if columns is not None else _gather_fieldnames(flat_records)
writer = csv.DictWriter(stream, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
if header_labels:
# DictWriter keys data rows by raw fieldname, so emit a relabeled header
# row manually (instead of writeheader) to keep header/data aligned.
csv.writer(stream).writerow([header_labels.get(f, f) for f in fieldnames])
else:
writer.writeheader()
for row in flat_records:
writer.writerow({k: ("" if v is None else v) for k, v in row.items()})

Expand Down
12 changes: 10 additions & 2 deletions nsc/output/render.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ def render(
*,
format: OutputFormat,
columns: list[str] | None = None,
header_labels: dict[str, str] | None = None,
stream: TextIO = sys.stdout,
compact: bool = False,
color: bool = False,
Expand All @@ -26,9 +27,16 @@ def render(
elif format is OutputFormat.YAML:
yaml_.render(data, stream=stream)
elif format is OutputFormat.CSV:
csv_.render(data, stream=stream, columns=columns)
csv_.render(data, stream=stream, columns=columns, header_labels=header_labels)
elif format is OutputFormat.TABLE:
table.render(data, stream=stream, columns=columns, color=color, object_colors=object_colors)
table.render(
data,
stream=stream,
columns=columns,
color=color,
object_colors=object_colors,
header_labels=header_labels,
)
else: # pragma: no cover (StrEnum exhaustively covered above)
raise ValueError(f"unknown output format: {format!r}")

Expand Down
3 changes: 2 additions & 1 deletion nsc/output/table.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ def render(
columns: list[str] | None = None,
color: bool = False,
object_colors: bool = False,
header_labels: dict[str, str] | None = None,
) -> None:
records = [data] if isinstance(data, dict) else list(data)
if not records:
Expand All @@ -48,7 +49,7 @@ def render(

table = Table(show_header=True, header_style="bold")
for col in fieldnames:
table.add_column(col)
table.add_column(header_labels.get(col, col) if header_labels else col)
for r in flat_records:
table.add_row(*[_format_cell(r.get(col, ""), color=color) for col in fieldnames])

Expand Down
151 changes: 151 additions & 0 deletions nsc/savedfilters/custom_fields.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
"""Resolve a model's custom-field definitions (name -> label, type, choices).

The list payload's ``custom_fields`` dict is keyed by field *name* only; the
human label and type live in the definitions at ``/api/extras/custom-fields/``,
filtered by ``object_type`` (e.g. ``dcim.device``). This mirrors
:class:`~nsc.savedfilters.objecttypes.ObjectTypeResolver`: fetch once per object
type, cache, and return ``None`` (never raise) when the registry can't be reached
so callers fall back to the raw key. Used to show custom-field column labels
(#132) and to build per-field edit widgets (#134).

Choice values for ``select``/``multiselect`` fields come from a referenced
choice set; they are fetched best-effort and left empty on any failure.
"""

from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any, Protocol

from nsc.http.errors import NetBoxAPIError, NetBoxClientError
from nsc.savedfilters.objecttypes import ObjectTypeResolver

_CUSTOM_FIELDS_PATH = "/api/extras/custom-fields/"
_CHOICE_SETS_PATH = "/api/extras/custom-field-choice-sets/"
_CF_PREFIX = "custom_fields."


class _ClientLike(Protocol):
def paginate(
self, path: str, params: dict[str, Any] | None = ..., *, limit: int | None = ...
) -> Any: ...

def get(self, path: str, params: dict[str, Any] | None = ...) -> dict[str, Any]: ...


@dataclass(frozen=True)
class CustomFieldDef:
name: str
label: str
type: str = "text"
choices: tuple[str, ...] = ()
required: bool = False


def humanize(name: str) -> str:
"""``site_contact`` -> ``Site Contact`` (fallback when a label is blank)."""
cleaned = name.replace("_", " ").strip()
return cleaned.title() if cleaned else name


def custom_field_labels(
columns: Iterable[str], defs: dict[str, CustomFieldDef] | None
) -> dict[str, str]:
"""Map every column key to its display label.

Non-custom-field keys map to themselves. ``custom_fields.<name>`` keys map to
the field's resolved label. When two visible custom-field columns resolve to
the same label they are ambiguous, so both fall back to their raw key. Unknown
custom fields and a missing ``defs`` also fall back to the raw key.
"""
cols = list(columns)
if not defs:
return {col: col for col in cols}
resolved: dict[str, str] = {}
for col in cols:
if col.startswith(_CF_PREFIX) and (cf := defs.get(col[len(_CF_PREFIX) :])) is not None:
resolved[col] = cf.label
seen: dict[str, list[str]] = {}
for col, label in resolved.items():
seen.setdefault(label, []).append(col)
labels: dict[str, str] = {}
for col in cols:
unique = resolved.get(col)
if unique is not None and len(seen[unique]) == 1:
labels[col] = unique
else:
labels[col] = col
return labels


class CustomFieldResolver:
"""Looks up custom-field definitions from the live API, cached per object type."""

def __init__(self, client: _ClientLike, object_types: ObjectTypeResolver | None = None) -> None:
self._client = client
self._object_types = object_types or ObjectTypeResolver(client)
self._cache: dict[str, dict[str, CustomFieldDef]] = {}
self._choice_cache: dict[int, tuple[str, ...]] = {}

def resolve(self, list_path: str) -> dict[str, CustomFieldDef] | None:
"""``{name: CustomFieldDef}`` for a list path, or ``None`` if unresolvable."""
object_type = self._object_types.resolve(list_path)
if object_type is None:
return None
if object_type in self._cache:
return self._cache[object_type]
try:
records = list(self._client.paginate(_CUSTOM_FIELDS_PATH, {"object_type": object_type}))
except (NetBoxAPIError, NetBoxClientError):
return None
defs = {d.name: d for d in (self._to_def(rec) for rec in records)}
self._cache[object_type] = defs
return defs

def _to_def(self, record: dict[str, Any]) -> CustomFieldDef:
name = str(record.get("name", ""))
raw_label = record.get("label")
label = raw_label if isinstance(raw_label, str) and raw_label else humanize(name)
cf_type = record.get("type")
type_value = cf_type.get("value") if isinstance(cf_type, dict) else cf_type
type_str = type_value if isinstance(type_value, str) else "text"
choices = self._choices_for(record.get("choice_set"))
return CustomFieldDef(
name=name,
label=label,
type=type_str,
choices=choices,
required=bool(record.get("required", False)),
)

def _choices_for(self, choice_set: Any) -> tuple[str, ...]:
if not isinstance(choice_set, dict):
return ()
cs_id = choice_set.get("id")
if not isinstance(cs_id, int):
return ()
if cs_id in self._choice_cache:
return self._choice_cache[cs_id]
try:
payload = self._client.get(f"{_CHOICE_SETS_PATH}{cs_id}/")
except (NetBoxAPIError, NetBoxClientError):
return ()
values = _extract_choice_values(payload)
self._choice_cache[cs_id] = values
return values


def _extract_choice_values(payload: dict[str, Any]) -> tuple[str, ...]:
raw = payload.get("choices")
if not isinstance(raw, list):
raw = payload.get("extra_choices")
if not isinstance(raw, list):
return ()
values: list[str] = []
for item in raw:
if isinstance(item, (list, tuple)) and item:
values.append(str(item[0]))
elif isinstance(item, str):
values.append(item)
return tuple(values)
62 changes: 62 additions & 0 deletions nsc/savedfilters/tags.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
"""Resolve the available NetBox tags for tag-picker widgets (#134).

Tags are global (not object-type scoped), so a single fetch of
``/api/extras/tags/`` is cached for the session. Mirrors the other resolvers:
returns ``None`` (never raises) when the API can't be reached, so the tag widget
degrades to free-text input. A writable ``tags`` PATCH wants a list of
``{name, slug}`` objects, so both are carried.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Protocol

from nsc.http.errors import NetBoxAPIError, NetBoxClientError
from nsc.output.colors import normalize_hex

_TAGS_PATH = "/api/extras/tags/"


class _ClientLike(Protocol):
def paginate(
self, path: str, params: dict[str, Any] | None = ..., *, limit: int | None = ...
) -> Any: ...


@dataclass(frozen=True)
class TagDef:
name: str
slug: str
color: str | None = None

@property
def label(self) -> str:
return self.name


class TagsResolver:
"""Fetches the global tag list once and caches it for the session."""

def __init__(self, client: _ClientLike) -> None:
self._client = client
self._cache: tuple[TagDef, ...] | None = None

def resolve(self) -> tuple[TagDef, ...] | None:
"""All tags, or ``None`` if the API can't be reached."""
if self._cache is not None:
return self._cache
try:
records = list(self._client.paginate(_TAGS_PATH))
except (NetBoxAPIError, NetBoxClientError):
return None
tags = tuple(
TagDef(
name=str(rec.get("name", "")),
slug=str(rec.get("slug", "")),
color=normalize_hex(rec.get("color")),
)
for rec in records
)
self._cache = tags
return tags
13 changes: 13 additions & 0 deletions nsc/tui/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,19 @@ def __init__(
self._saved_filter_store = saved_filter_store
if saved_filter_store is not None and getattr(saved_filter_store, "on_error", None) is None:
saved_filter_store.on_error = self._notify_saved_filter_issue
from nsc.savedfilters.custom_fields import CustomFieldResolver # noqa: PLC0415
from nsc.savedfilters.tags import TagsResolver # noqa: PLC0415

self._custom_field_resolver = CustomFieldResolver(client)
self._tags_resolver = TagsResolver(client)

def custom_field_defs_for(self, tag: str, resource: str) -> dict[str, Any] | None:
"""Custom-field definitions for a resource (read by list/forms), or None."""
return self._custom_field_resolver.resolve(self._list_path(tag, resource))

def available_tags(self) -> Any | None:
"""All NetBox tags (read by the tag-picker widget), or None if unavailable."""
return self._tags_resolver.resolve()

def columns_for(self, tag: str, resource: str) -> list[str] | None:
"""Saved visible columns for a resource, if any (read by ListScreen)."""
Expand Down
Loading