Commit 277bbda8 authored by Jan Reimes's avatar Jan Reimes
Browse files

refactor(02-06): convert Workspace and WorkspaceMember to dataclasses 

- Converted Workspace from BaseModel to @dataclass
- Converted WorkspaceMember from BaseModel to @dataclass
- Validation logic moved from @field_validator to __post_init__
- GraphNode and GraphEdge remain as BaseModel (LightRAG requirement)
- Updated imports in workspace_registry.py to use models.WorkspaceMember
- make_workspace_member now uses models.WorkspaceMember for consistency
- Maintains backward compatibility with existing code
parent f69b9c33

packages/3gpp-ai/threegpp_ai/models.py

+28 −30
Original line number Diff line number Diff line
@@ -2,6 +2,7 @@ 


from __future__ import annotations



from dataclasses import dataclass, field

from datetime import datetime

from enum import StrEnum, auto

from typing import Any
@@ -79,51 +80,48 @@ class WorkspaceNotFoundError(AiError): 
    """Workspace does not exist in registry."""





class Workspace(BaseModel):

@dataclass

class Workspace:

    """Logical workspace boundary for AI processing."""



    workspace_name: str = Field(..., description="Normalized workspace identifier")

    created_at: datetime = Field(default_factory=utc_now, description="Creation timestamp")

    updated_at: datetime = Field(default_factory=utc_now, description="Last update timestamp")

    is_default: bool = Field(False, description="Whether this workspace is the default workspace")

    auto_build: bool | None = Field(default=False, description="Whether to auto-build embeddings on TDoc add")

    workspace_name: str = field(metadata={"description": "Normalized workspace identifier"})

    created_at: datetime = field(default_factory=utc_now, metadata={"description": "Creation timestamp"})

    updated_at: datetime = field(default_factory=utc_now, metadata={"description": "Last update timestamp"})

    is_default: bool = field(default=False, metadata={"description": "Whether this workspace is the default workspace"})

    auto_build: bool | None = field(default=False, metadata={"description": "Whether to auto-build embeddings on TDoc add"})



    @field_validator("workspace_name")

    @classmethod

    def _normalize_workspace_name(cls, value: str) -> str:

        if not value.strip():

    def __post_init__(self) -> None:

        """Normalize workspace_name after initialization."""

        if not self.workspace_name.strip():

            msg = "workspace_name must not be empty"

            raise ValueError(msg)

        return normalize_workspace_name(value)

        self.workspace_name = normalize_workspace_name(self.workspace_name)





class WorkspaceMember(BaseModel):

@dataclass

class WorkspaceMember:

    """Source item assigned to one workspace corpus."""



    workspace_name: str = Field(..., description="Workspace identifier")

    source_item_id: str = Field(..., description="Stable source item identifier")

    source_path: str = Field(..., description="Path or locator of the source item")

    source_kind: SourceKind = Field(..., description="Type of source item")

    added_at: datetime = Field(default_factory=utc_now, description="Registration timestamp")

    added_by: str | None = Field(None, description="Actor that registered the source")

    is_active: bool = Field(True, description="Membership active flag")



    @field_validator("workspace_name")

    @classmethod

    def _normalize_workspace_name(cls, value: str) -> str:

        if not value.strip():

    workspace_name: str = field(metadata={"description": "Workspace identifier"})

    source_item_id: str = field(metadata={"description": "Stable source item identifier"})

    source_path: str = field(metadata={"description": "Path or locator of the source item"})

    source_kind: SourceKind = field(metadata={"description": "Type of source item"})

    added_at: datetime = field(default_factory=utc_now, metadata={"description": "Registration timestamp"})

    added_by: str | None = field(default=None, metadata={"description": "Actor that registered the source"})

    is_active: bool = field(default=True, metadata={"description": "Membership active flag"})



    def __post_init__(self) -> None:

        """Normalize fields after initialization."""

        if not self.workspace_name.strip():

            msg = "workspace_name must not be empty"

            raise ValueError(msg)

        return normalize_workspace_name(value)

        self.workspace_name = normalize_workspace_name(self.workspace_name)



    @field_validator("source_item_id")

    @classmethod

    def _normalize_source_item_id(cls, value: str) -> str:

        normalized = normalize_tdoc_id(value)

        normalized = normalize_tdoc_id(self.source_item_id)

        if not normalized:

            msg = "source_item_id must not be empty"

            raise ValueError(msg)

        return normalized

        self.source_item_id = normalized





class DocumentClassification(BaseModel):

packages/3gpp-ai/threegpp_ai/operations/workspaces.py

+4 −6
Original line number Diff line number Diff line
@@ -21,12 +21,9 @@ from tdoc_crawler.tdocs.sources.portal import PortalSource 
from tdoc_crawler.tdocs.sources.whatthespec import resolve_via_whatthespec

from tdoc_crawler.utils.normalization import normalize_spec_number, normalize_tdoc_id, resolve_release_to_full_version



from threegpp_ai.models import WorkspaceNotFoundError

from threegpp_ai.models import WorkspaceMember, WorkspaceNotFoundError

from threegpp_ai.operations.workspace_names import DEFAULT_WORKSPACE, is_default_workspace, normalize_workspace_name

from threegpp_ai.operations.workspace_registry import (

    WorkspaceMember,

    WorkspaceRegistry,

)

from threegpp_ai.operations.workspace_registry import WorkspaceRegistry



_logger = get_logger(__name__)
@@ -384,8 +381,9 @@ def make_workspace_member( 
    Returns:

        WorkspaceMember instance.

    """

    normalize_workspace_name(workspace)

    normalized_workspace = normalize_workspace_name(workspace) if workspace else ""

    return WorkspaceMember(

        workspace_name=normalized_workspace,

        source_item_id=source_item_id,

        source_path=str(source_path),

        source_kind=source_kind,