204 lines
6.7 KiB
Python
204 lines
6.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Iterable, List, Sequence
|
|
|
|
from fastapi import Request
|
|
|
|
from models.navigation import NavigationLink
|
|
from services.repositories import NavigationRepository
|
|
from services.session import AuthSession
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class NavigationLinkDTO:
|
|
id: int
|
|
label: str
|
|
href: str
|
|
match_prefix: str | None
|
|
icon: str | None
|
|
tooltip: str | None
|
|
is_external: bool
|
|
children: List["NavigationLinkDTO"] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class NavigationGroupDTO:
|
|
id: int
|
|
label: str
|
|
icon: str | None
|
|
tooltip: str | None
|
|
links: List[NavigationLinkDTO] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class NavigationSidebarDTO:
|
|
groups: List[NavigationGroupDTO]
|
|
roles: tuple[str, ...]
|
|
|
|
|
|
class NavigationService:
|
|
"""Build navigation payloads filtered for the current session."""
|
|
|
|
def __init__(self, repository: NavigationRepository) -> None:
|
|
self._repository = repository
|
|
|
|
def build_sidebar(
|
|
self,
|
|
*,
|
|
session: AuthSession,
|
|
request: Request | None = None,
|
|
include_disabled: bool = False,
|
|
) -> NavigationSidebarDTO:
|
|
roles = self._collect_roles(session)
|
|
groups = self._repository.list_groups_with_links(
|
|
include_disabled=include_disabled
|
|
)
|
|
context = self._derive_context(request)
|
|
|
|
mapped_groups: List[NavigationGroupDTO] = []
|
|
for group in groups:
|
|
if not include_disabled and not group.is_enabled:
|
|
continue
|
|
mapped_links = self._map_links(
|
|
group.links,
|
|
roles,
|
|
request=request,
|
|
include_disabled=include_disabled,
|
|
context=context,
|
|
)
|
|
if not mapped_links and not include_disabled:
|
|
continue
|
|
mapped_groups.append(
|
|
NavigationGroupDTO(
|
|
id=group.id,
|
|
label=group.label,
|
|
icon=group.icon,
|
|
tooltip=group.tooltip,
|
|
links=mapped_links,
|
|
)
|
|
)
|
|
return NavigationSidebarDTO(groups=mapped_groups, roles=roles)
|
|
|
|
def _map_links(
|
|
self,
|
|
links: Sequence[NavigationLink],
|
|
roles: Iterable[str],
|
|
*,
|
|
request: Request | None,
|
|
include_disabled: bool,
|
|
context: dict[str, str | None],
|
|
include_children: bool = False,
|
|
) -> List[NavigationLinkDTO]:
|
|
resolved_roles = tuple(roles)
|
|
mapped: List[NavigationLinkDTO] = []
|
|
for link in sorted(links, key=lambda x: (x.sort_order, x.id)):
|
|
if not include_children and link.parent_link_id is not None:
|
|
continue
|
|
if not include_disabled and (not link.is_enabled):
|
|
continue
|
|
if not self._link_visible(link, resolved_roles, include_disabled):
|
|
continue
|
|
href = self._resolve_href(link, request=request, context=context)
|
|
if not href:
|
|
continue
|
|
children = self._map_links(
|
|
link.children,
|
|
resolved_roles,
|
|
request=request,
|
|
include_disabled=include_disabled,
|
|
context=context,
|
|
include_children=True,
|
|
)
|
|
match_prefix = link.match_prefix or href
|
|
mapped.append(
|
|
NavigationLinkDTO(
|
|
id=link.id,
|
|
label=link.label,
|
|
href=href,
|
|
match_prefix=match_prefix,
|
|
icon=link.icon,
|
|
tooltip=link.tooltip,
|
|
is_external=link.is_external,
|
|
children=children,
|
|
)
|
|
)
|
|
return mapped
|
|
|
|
@staticmethod
|
|
def _collect_roles(session: AuthSession) -> tuple[str, ...]:
|
|
roles = tuple((session.role_slugs or ()) if session else ())
|
|
if session and session.is_authenticated:
|
|
return roles
|
|
if "anonymous" in roles:
|
|
return roles
|
|
return roles + ("anonymous",)
|
|
|
|
@staticmethod
|
|
def _derive_context(request: Request | None) -> dict[str, str | None]:
|
|
if request is None:
|
|
return {"project_id": None, "scenario_id": None}
|
|
project_id = request.path_params.get(
|
|
"project_id") if hasattr(request, "path_params") else None
|
|
scenario_id = request.path_params.get(
|
|
"scenario_id") if hasattr(request, "path_params") else None
|
|
if not project_id:
|
|
project_id = request.query_params.get("project_id")
|
|
if not scenario_id:
|
|
scenario_id = request.query_params.get("scenario_id")
|
|
return {"project_id": project_id, "scenario_id": scenario_id}
|
|
|
|
def _resolve_href(
|
|
self,
|
|
link: NavigationLink,
|
|
*,
|
|
request: Request | None,
|
|
context: dict[str, str | None],
|
|
) -> str | None:
|
|
if link.route_name:
|
|
if request is None:
|
|
fallback = link.href_override
|
|
if fallback:
|
|
return fallback
|
|
# Fallback to route name when no request is available
|
|
return f"/{link.route_name.replace('.', '/')}"
|
|
requires_context = link.slug in {
|
|
"profitability",
|
|
"profitability-calculator",
|
|
"opex",
|
|
"capex",
|
|
}
|
|
if requires_context:
|
|
project_id = context.get("project_id")
|
|
scenario_id = context.get("scenario_id")
|
|
if project_id and scenario_id:
|
|
try:
|
|
return str(
|
|
request.url_for(
|
|
link.route_name,
|
|
project_id=project_id,
|
|
scenario_id=scenario_id,
|
|
)
|
|
)
|
|
except Exception: # pragma: no cover - defensive
|
|
pass
|
|
try:
|
|
return str(request.url_for(link.route_name))
|
|
except Exception: # pragma: no cover - defensive
|
|
return link.href_override
|
|
return link.href_override
|
|
|
|
@staticmethod
|
|
def _link_visible(
|
|
link: NavigationLink,
|
|
roles: Iterable[str],
|
|
include_disabled: bool,
|
|
) -> bool:
|
|
role_tuple = tuple(roles)
|
|
if not include_disabled and not link.is_enabled:
|
|
return False
|
|
if not link.required_roles:
|
|
return True
|
|
role_set = set(role_tuple)
|
|
return any(role in role_set for role in link.required_roles)
|