orchestrata_ultimate.knowledge_graph.portfolio

Reference › orchestrata_ultimate › knowledge_graph › portfolio

KGPortfolioManager manages portfolios (named groups of projects) for the knowledge graph package, traced to Spec 008 REQ-008-003, -007 and -009. The constructor optionally takes a graph_client GraphClient. The operations shown in the evidence keep their state in in-process dictionaries keyed by a generated UUID: create_portfolio, add_project, get_portfolio, get_project, get_portfolio_projects, remove_project, plus the portfolio_count and project_count properties. get_dashboard builds a Dashboard (from types) holding the portfolio id and its project count, and cross_project_query searches an entity name across a portfolio's projects. Because the dictionaries live in the process, portfolios do not survive a restart. No other module imports it; four tests cover it.

At a glance

Class diagram

Dependency diagram

API

Classes

KGPortfolioManager

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-003', 'REQ-008-007']) · Line 31

Manages portfolios and projects in the knowledge graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/portfolio.py

# mypy: ignore-errors
"""Portfolio management for the knowledge graph.

Manages portfolio-based organization, cross-project context,
and dashboard generation.
"""

from __future__ import annotations

import logging
import uuid
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.types import Dashboard

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

logger = logging.getLogger(__name__)


@traces(
    spec="008",
    story="US-001",
    reqs=[
        "REQ-008-003",
        "REQ-008-007",
    ],
)
class KGPortfolioManager:
    """Manages portfolios and projects in the knowledge graph."""

    def __init__(self, graph_client: GraphClient | None = None) -> None:
        self._client = graph_client
        self._portfolios: dict[str, dict[str, Any]] = {}
        self._projects: dict[str, dict[str, Any]] = {}
        self._portfolio_projects: dict[str, list[str]] = {}

    @traces(spec="008", story="US-001", reqs=["REQ-008-007"])
    def create_portfolio(
        self, name: str, metadata: dict[str, Any] | None = None
    ) -> str:
        """Create a new portfolio. Returns portfolio ID."""
        portfolio_id = str(uuid.uuid4())
        self._portfolios[portfolio_id] = {
            "id": portfolio_id,
            "name": name,
            "metadata": metadata or {},
        }
        self._portfolio_projects[portfolio_id] = []
        return portfolio_id

    @traces(spec="008", story="US-001", reqs=["REQ-008-007"])
    def add_project(
        self, portfolio_id: str, project_name: str, project_path: str
    ) -> str | None:
        """Add a project to a portfolio. Returns project ID."""
        if portfolio_id not in self._portfolios:
            return None
        project_id = str(uuid.uuid4())
        self._projects[project_id] = {
            "id": project_id,
            "name": project_name,
            "path": project_path,
            "portfolio_id": portfolio_id,
        }
        self._portfolio_projects[portfolio_id].append(project_id)
        return project_id

    def get_portfolio(self, portfolio_id: str) -> dict[str, Any] | None:
        """Get portfolio details."""
        return self._portfolios.get(portfolio_id)

    def get_project(self, project_id: str) -> dict[str, Any] | None:
        """Get project details."""
        return self._projects.get(project_id)

    @traces(spec="008", story="US-001", reqs=["REQ-008-048"])
    def get_portfolio_projects(self, portfolio_id: str) -> list[dict[str, Any]]:
        """Get all projects in a portfolio."""
        project_ids = self._portfolio_projects.get(portfolio_id, [])
        return [self._projects[pid] for pid in project_ids if pid in self._projects]

    @traces(spec="008", story="US-001", reqs=["REQ-008-009"])
    def get_dashboard(self, portfolio_id: str) -> Dashboard:
        """Generate a dashboard for a portfolio."""
        projects = self.get_portfolio_projects(portfolio_id)
        return Dashboard(
            portfolio_id=portfolio_id,
            project_count=len(projects),
        )

    @traces(spec="008", story="US-001", reqs=["REQ-008-003"])
    def cross_project_query(
        self, portfolio_id: str, entity_name: str
    ) -> list[dict[str, Any]]:
        """Search for an entity across all projects in a portfolio."""
        projects = self.get_portfolio_projects(portfolio_id)
        return [
            {
                "project_id": project["id"],
                "project_name": project["name"],
                "entity_name": entity_name,
            }
            for project in projects
        ]

    def remove_project(self, project_id: str) -> bool:
        """Remove a project from its portfolio."""
        project = self._projects.get(project_id)
        if not project:
            return False
        portfolio_id = project.get("portfolio_id", "")
        if portfolio_id in self._portfolio_projects:
            self._portfolio_projects[portfolio_id] = [
                p for p in self._portfolio_projects[portfolio_id] if p != project_id
            ]
        del self._projects[project_id]
        return True

    @property
    def portfolio_count(self) -> int:
        return len(self._portfolios)

    @property
    def project_count(self) -> int:
        return len(self._projects)