Create store.py

api/store.py

@@ -0,0 +1,145 @@
+# api/store.py
+from __future__ import annotations
+import json
+import os
+from typing import Dict, Any, List, Optional
+from pathlib import Path
+
+from api.models import CourseDirectoryItem, Workspace, PersonContact, WorkspaceCourseRef, WorkspaceMember
+
+DATA_DIR = Path(os.getenv("CLARE_DATA_DIR", "./data"))
+COURSE_FILE = DATA_DIR / "course_directory.json"
+WORKSPACE_FILE = DATA_DIR / "workspaces.json"
+
+
+def _ensure_dir():
+  DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _read_json(path: Path) -> Optional[Any]:
+  if not path.exists():
+    return None
+  with path.open("r", encoding="utf-8") as f:
+    return json.load(f)
+
+
+def _write_json(path: Path, obj: Any) -> None:
+  _ensure_dir()
+  with path.open("w", encoding="utf-8") as f:
+    json.dump(obj, f, ensure_ascii=False, indent=2)
+
+
+def seed_if_missing() -> None:
+  """
+  Provide sane defaults so HF can run out-of-the-box.
+  Replace these with your real course list/workspaces later.
+  """
+  _ensure_dir()
+
+  if not COURSE_FILE.exists():
+    courses = [
+      CourseDirectoryItem(
+        id="course_ai_001",
+        name="Introduction to AI",
+        instructor=PersonContact(name="Dr. Smith", email="smith@uni.edu"),
+        teachingAssistant=PersonContact(name="Alice", email="alice@uni.edu"),
+      ).model_dump()
+    ]
+    _write_json(COURSE_FILE, {"items": courses})
+
+  if not WORKSPACE_FILE.exists():
+    workspaces = [
+      Workspace(
+        id="ws_group_ai_g3",
+        type="group",
+        category="course",
+        name="Group Alpha",
+        groupNo=3,
+        courseId="course_ai_001",
+        courseInfo=WorkspaceCourseRef(id="course_ai_001", name="Introduction to AI"),
+        members=[
+          WorkspaceMember(id="u1", name="Sarah", email="sarah@uni.edu", role="owner"),
+          WorkspaceMember(id="u2", name="Bob", email="bob@uni.edu", role="member"),
+        ],
+      ).model_dump()
+    ]
+    _write_json(WORKSPACE_FILE, {"items": workspaces})
+
+
+def load_courses() -> List[CourseDirectoryItem]:
+  seed_if_missing()
+  raw = _read_json(COURSE_FILE) or {}
+  items = raw.get("items", [])
+  out: List[CourseDirectoryItem] = []
+  for x in items:
+    out.append(CourseDirectoryItem(**x))
+  return out
+
+
+def save_courses(courses: List[CourseDirectoryItem]) -> None:
+  _write_json(COURSE_FILE, {"items": [c.model_dump() for c in courses]})
+
+
+def load_workspaces() -> List[Workspace]:
+  seed_if_missing()
+  raw = _read_json(WORKSPACE_FILE) or {}
+  items = raw.get("items", [])
+  out: List[Workspace] = []
+  for x in items:
+    out.append(Workspace(**x))
+  return out
+
+
+def save_workspaces(workspaces: List[Workspace]) -> None:
+  _write_json(WORKSPACE_FILE, {"items": [w.model_dump() for w in workspaces]})
+
+
+def enrich_workspace_course_info(workspaces: List[Workspace], courses: List[CourseDirectoryItem]) -> List[Workspace]:
+  """
+  Ensure workspace.courseInfo has instructor/TA to satisfy CourseInfoSection fallback.
+  Frontend prefers availableCourses, but this makes fallback robust.
+  """
+  by_id = {c.id.strip().lower(): c for c in courses}
+  by_name = {c.name.strip().lower(): c for c in courses}
+
+  enriched: List[Workspace] = []
+  for w in workspaces:
+    ws_course = w.courseInfo
+    hit: Optional[CourseDirectoryItem] = None
+
+    if w.courseId:
+      hit = by_id.get(w.courseId.strip().lower())
+    if not hit and ws_course and ws_course.id:
+      hit = by_id.get(ws_course.id.strip().lower())
+    if not hit and ws_course and ws_course.name:
+      hit = by_name.get(ws_course.name.strip().lower())
+
+    if hit:
+      w.courseId = w.courseId or hit.id
+      w.courseInfo = WorkspaceCourseRef(
+        id=hit.id,
+        name=hit.name,
+        instructor=hit.instructor,
+        teachingAssistant=hit.teachingAssistant,
+      )
+    else:
+      # keep whatever exists; CourseInfoSection will show fallback if nothing matches
+      w.courseInfo = ws_course
+
+    enriched.append(w)
+
+  return enriched
+
+
+def rename_workspace(workspace_id: str, new_name: str) -> Workspace:
+  workspaces = load_workspaces()
+  found = None
+  for w in workspaces:
+    if w.id == workspace_id:
+      w.name = new_name
+      found = w
+      break
+  if not found:
+    raise KeyError(f"workspace not found: {workspace_id}")
+  save_workspaces(workspaces)
+  return found