WorkspaceService

WorkspaceService manages the lifecycle of workspaces for the current application session. Each workspace maps to a directory on disk containing a dedicated SQLite database file.

Session-Scoped Registry

The service maintains an in-memory registry of Workspace and SqliteGraphRepository objects for the current session. Workspaces created in previous sessions are not automatically reloaded — this will be addressed in a future persistence pass.

Default Storage Directory

If no base_directory is specified, workspaces are stored under ~/.knowledge_platform/:

from knowledge_platform.services.workspace_service import WorkspaceService

svc = WorkspaceService()                         # uses ~/.knowledge_platform/
svc = WorkspaceService(base_directory=Path("/data/workspaces"))  # custom path

Creating and Accessing Workspaces

# Create
workspace = svc.create_workspace("My Research Notes")
print(workspace.db_path)  # Path("~/.knowledge_platform/<uuid>.db")

# Retrieve
ws = svc.get_workspace(workspace.id)

# List all workspaces in the current session
all_ws = svc.list_workspaces()

Getting a Repository

Each workspace has exactly one SqliteGraphRepository. Retrieve it to pass to a GraphEngine:

from knowledge_platform.core.engine import GraphEngine

repo = svc.get_repository(workspace.id)
engine = GraphEngine(repo)

Cleanup

Call close() when the service is being torn down to release all SQLite connection pools:

svc.close()  # disposes all repositories

In the standard application flow, WorkspaceService.close() is called at application exit. In tests, use a yield-based fixture:

@pytest.fixture()
def workspace_service(tmp_path):
    svc = WorkspaceService(base_directory=tmp_path)
    yield svc
    svc.close()

API Reference

knowledge_platform.services.workspace_service.WorkspaceService

Manages workspace creation, loading, and listing.

Workspaces are persisted as directories containing a SQLite database. The service keeps an in-memory registry of open workspaces for the current session.

Parameters:

Name Type Description Default
base_directory Path | None

Parent directory where workspace databases are stored. Defaults to ~/.knowledge_platform.

 None
Source code in src/knowledge_platform/services/workspace_service.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class WorkspaceService:
    """Manages workspace creation, loading, and listing.

    Workspaces are persisted as directories containing a SQLite database.
    The service keeps an in-memory registry of open workspaces for the
    current session.

    Args:
        base_directory: Parent directory where workspace databases are stored.
            Defaults to ``~/.knowledge_platform``.
    """

    def __init__(self, base_directory: Path | None = None) -> None:
        if base_directory is None:
            base_directory = Path.home() / ".knowledge_platform"
        base_directory.mkdir(parents=True, exist_ok=True)
        self._base_directory = base_directory
        self._workspaces: dict[WorkspaceId, Workspace] = {}
        self._repositories: dict[WorkspaceId, SqliteGraphRepository] = {}

    def _register_workspace(self, workspace: Workspace, repo: SqliteGraphRepository) -> Workspace:
        self._workspaces[workspace.id] = workspace
        self._repositories[workspace.id] = repo
        return workspace

    def _discover_workspaces(self) -> None:
        for db_path in sorted(self._base_directory.glob("*.db")):
            if any(workspace.db_path == db_path for workspace in self._workspaces.values()):
                continue
            repo = SqliteGraphRepository(db_path)
            workspace = repo.load_workspace()
            if workspace is None:
                repo.close()
                continue
            self._register_workspace(workspace, repo)

    def create_workspace(self, name: str) -> Workspace:
        """Create and register a new workspace.

        Args:
            name: Display name.

        Returns:
            Newly created :class:`~knowledge_platform.workspace.Workspace`.
        """
        workspace = Workspace.create(name, self._base_directory)
        repo = SqliteGraphRepository(workspace.db_path)
        repo.ensure_workspace(workspace.id, name)
        self._register_workspace(workspace, repo)
        logger.info("service.workspace.created", workspace_id=workspace.id, name=name)
        return workspace

    def open_workspace(self, workspace_id: WorkspaceId) -> Workspace:
        """Load an existing workspace from disk into the current session."""
        self._discover_workspaces()
        return self.get_workspace(workspace_id)

    def get_workspace(self, workspace_id: WorkspaceId) -> Workspace:
        """Return a registered workspace.

        Args:
            workspace_id: Target identifier.

        Returns:
            The :class:`~knowledge_platform.workspace.Workspace`.

        Raises:
            KeyError: If not registered in the current session.
        """
        self._discover_workspaces()
        if workspace_id not in self._workspaces:
            raise KeyError(f"Workspace {workspace_id!r} is not registered in this session.")
        return self._workspaces[workspace_id]

    def get_repository(self, workspace_id: WorkspaceId) -> SqliteGraphRepository:
        """Return the SQLite repository for a workspace.

        Args:
            workspace_id: Target identifier.

        Returns:
            The :class:`~knowledge_platform.persistence.SqliteGraphRepository`.

        Raises:
            KeyError: If workspace not registered.
        """
        self._discover_workspaces()
        if workspace_id not in self._repositories:
            raise KeyError(f"No repository for workspace {workspace_id!r}.")
        return self._repositories[workspace_id]

    def list_workspaces(self) -> list[Workspace]:
        """Return all currently registered workspaces.

        Returns:
            List of :class:`~knowledge_platform.workspace.Workspace` instances.
        """
        self._discover_workspaces()
        return list(self._workspaces.values())

    def close(self) -> None:
        """Dispose all open repository connection pools.

        Call this when the service is being torn down (e.g. application exit
        or end of a test) to release SQLite file handles cleanly.
        """
        for repo in self._repositories.values():
            repo.close()
        self._repositories.clear()

Functions

close 
close() -> None

Dispose all open repository connection pools.

Call this when the service is being torn down (e.g. application exit or end of a test) to release SQLite file handles cleanly.

Source code in src/knowledge_platform/services/workspace_service.py 
116
117
118
119
120
121
122
123
124
def close(self) -> None:
    """Dispose all open repository connection pools.

    Call this when the service is being torn down (e.g. application exit
    or end of a test) to release SQLite file handles cleanly.
    """
    for repo in self._repositories.values():
        repo.close()
    self._repositories.clear()
create_workspace 
create_workspace(name: str) -> Workspace

Create and register a new workspace.

Parameters:

Name Type Description Default
name str

Display name.

 required

Returns:

Type Description
Workspace

Newly created :class:~knowledge_platform.workspace.Workspace.
Source code in src/knowledge_platform/services/workspace_service.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def create_workspace(self, name: str) -> Workspace:
    """Create and register a new workspace.

    Args:
        name: Display name.

    Returns:
        Newly created :class:`~knowledge_platform.workspace.Workspace`.
    """
    workspace = Workspace.create(name, self._base_directory)
    repo = SqliteGraphRepository(workspace.db_path)
    repo.ensure_workspace(workspace.id, name)
    self._register_workspace(workspace, repo)
    logger.info("service.workspace.created", workspace_id=workspace.id, name=name)
    return workspace
get_repository 
get_repository(
    workspace_id: WorkspaceId,
) -> SqliteGraphRepository

Return the SQLite repository for a workspace.

Parameters:

Name Type Description Default
workspace_id WorkspaceId

Target identifier.

 required

Returns:

Name Type Description
The SqliteGraphRepository

class:~knowledge_platform.persistence.SqliteGraphRepository.

Raises:

Type Description
KeyError

If workspace not registered.
Source code in src/knowledge_platform/services/workspace_service.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def get_repository(self, workspace_id: WorkspaceId) -> SqliteGraphRepository:
    """Return the SQLite repository for a workspace.

    Args:
        workspace_id: Target identifier.

    Returns:
        The :class:`~knowledge_platform.persistence.SqliteGraphRepository`.

    Raises:
        KeyError: If workspace not registered.
    """
    self._discover_workspaces()
    if workspace_id not in self._repositories:
        raise KeyError(f"No repository for workspace {workspace_id!r}.")
    return self._repositories[workspace_id]
get_workspace 
get_workspace(workspace_id: WorkspaceId) -> Workspace

Return a registered workspace.

Parameters:

Name Type Description Default
workspace_id WorkspaceId

Target identifier.

 required

Returns:

Name Type Description
The Workspace

class:~knowledge_platform.workspace.Workspace.

Raises:

Type Description
KeyError

If not registered in the current session.
Source code in src/knowledge_platform/services/workspace_service.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def get_workspace(self, workspace_id: WorkspaceId) -> Workspace:
    """Return a registered workspace.

    Args:
        workspace_id: Target identifier.

    Returns:
        The :class:`~knowledge_platform.workspace.Workspace`.

    Raises:
        KeyError: If not registered in the current session.
    """
    self._discover_workspaces()
    if workspace_id not in self._workspaces:
        raise KeyError(f"Workspace {workspace_id!r} is not registered in this session.")
    return self._workspaces[workspace_id]
list_workspaces 
list_workspaces() -> list[Workspace]

Return all currently registered workspaces.

Returns:

Type Description
list[Workspace]

List of :class:~knowledge_platform.workspace.Workspace instances.
Source code in src/knowledge_platform/services/workspace_service.py 
107
108
109
110
111
112
113
114
def list_workspaces(self) -> list[Workspace]:
    """Return all currently registered workspaces.

    Returns:
        List of :class:`~knowledge_platform.workspace.Workspace` instances.
    """
    self._discover_workspaces()
    return list(self._workspaces.values())
open_workspace 
open_workspace(workspace_id: WorkspaceId) -> Workspace

Load an existing workspace from disk into the current session.

Source code in src/knowledge_platform/services/workspace_service.py 
68
69
70
71
def open_workspace(self, workspace_id: WorkspaceId) -> Workspace:
    """Load an existing workspace from disk into the current session."""
    self._discover_workspaces()
    return self.get_workspace(workspace_id)