feat: add InternalModel for non-queryable models

campus/api/resources/timetable.py

@@ -0,0 +1,120 @@
+"""campus.api.resources.submission
+
+Timetable resource for Campus API.
+"""
+
+import typing
+from campus.common import schema
+from campus.common.errors import api_errors
+from campus.common.utils import uid
+import campus.model
+import campus.storage
+
+timetable_storage = campus.storage.get_collection("timetables")
+
+def _from_record(record: dict) -> campus.model.Timetable:
+    """Convert storage record to Timetable model."""
+    return campus.model.Timetable(
+        id=schema.CampusID(record["id"]),
+        filename=record["filename"],
+        lessongroup_id=record["lessongroup_id"],
+        venuetimeslot_id=record["venuetimeslot_id"],
+        created_at=schema.DateTime(record["created_at"]) if record.get("created_at") else None,
+        updated_at=schema.DateTime(record["updated_at"]) if record.get("updated_at") else None,
+    )
+
+class TimetablesResource:
+    """Represents the timetables resource."""
+
+    @staticmethod
+    def init_storage() -> None:
+        """Initialize storage."""
+        timetable_storage.init_collection()
+
+    def __getitem__(self, timetable_id: schema.CampusID) -> "TimetableResource":
+        return TimetableResource(timetable_id)
+
+    def list(self, **filters: typing.Any) -> list[campus.model.Timetable]:
+        """List timetables matching filters."""
+        try:
+            records = timetable_storage.get_matching(filters)
+        except campus.storage.errors.StorageError as e:
+            raise api_errors.InternalError.from_exception(e) from e
+        return [_from_record(record) for record in records]
+
+    def new(self, **fields: typing.Any) -> campus.model.Timetable:
+        """Create new timetable."""
+
+        timetable = campus.model.Timetable(
+            id = schema.CampusID(
+                uid.generate_category_uid("timetable", length=8)
+            ),
+            filename=fields["filename"],
+            lessongroup_id = fields["lessongroup_id"],
+            venuetimeslot_id = fields["venuetimeslot_id"],
+            created_at=schema.DateTime.utcnow()
+        )
+
+        try:
+            timetable_storage.insert_one(timetable.to_storage())
+        except campus.storage.errors.StorageError as e:
+            raise api_errors.InternalError.from_exception(e) from e
+
+        return timetable
+
+class TimetableResource:
+    """Represents a single timetable."""
+
+    def __init__(self, timetable_id: schema.CampusID):
+        self.timetable_id = timetable_id
+
+    def get(self) -> campus.model.Timetable:
+        """Get the timetable."""
+        try:
+            record = timetable_storage.get_by_id(self.timetable_id)
+            if record is None:
+                raise api_errors.ConflictError(
+                    "Timetable not found",
+                    id=self.timetable_id
+                )
+            return _from_record(record)
+        except campus.storage.errors.NotFoundError:
+            raise api_errors.ConflictError(
+                "Timetable not found",
+                id=self.timetable_id
+            ) from None
+        except campus.storage.errors.StorageError as e:
+            raise api_errors.InternalError.from_exception(e) from e
+
+
+    def update(self, **updates: typing.Any) -> None:
+        """Update the timetable record."""
+        try:
+            timetable_storage.update_by_id(self.timetable_id, updates)
+        except campus.storage.errors.NoChangesAppliedError:
+            return None
+        except campus.storage.errors.NotFoundError:
+            raise api_errors.ConflictError(
+                "Timetable not found",
+                id=self.timetable_id
+            ) from None
+        except campus.storage.errors.StorageError as e:
+            raise api_errors.InternalError.from_exception(e) from e
+
+    def delete(self) -> None:
+        """Delete the timetable."""
+        try:
+            record = timetable_storage.get_by_id(self.timetable_id)
+            if record is None:
+                raise api_errors.ConflictError(
+                    "Timetable not found",
+                    id=self.timetable_id
+                )
+            timetable_storage.delete_by_id(self.timetable_id)
+        except campus.storage.errors.NotFoundError:
+            raise api_errors.ConflictError(
+                "Timetable not found",
+                id=self.timetable_id
+            ) from None
+        except campus.storage.errors.StorageError as e:
+            raise api_errors.InternalError.from_exception(e) from e

campus/model/__init__.py

@@ -25,6 +25,7 @@
     "Feedback",
     "HttpHeader",
     "HttpHeaderWithAuth",
+    "InternalModel",
     "LoginSession",
     "Model",
     "Question",
@@ -37,7 +38,7 @@
 ]
 
 from .assignment import Assignment, ClassroomLink, Question
-from .base import Model
+from .base import InternalModel, Model
 from .circle import Circle
 from .client import Client, ClientAccess
 from .credentials import OAuthToken, UserCredentials

campus/model/base.py

@@ -26,9 +26,26 @@ class FieldMeta(typing.TypedDict):
     constraints: typing.Sequence[str]
 
 
+@dataclass(kw_only=True)
+class InternalModel(typing.Protocol):
+    """Base class for internal models in Campus.
+
+    Internal models are not exposed through Campus API endpoints,
+    but are used internally as intermediate representations.
+    """
+
+    @classmethod
+    def fields(cls) -> dict[str, dataclasses.Field]:  # type: ignore[override]
+        return {field.name: field for field in dataclasses.fields(cls)}
+
+
 @dataclass(kw_only=True)
 class Model(typing.Protocol):
-    """Base class for all models in Campus."""
+    """Base class for all public models in Campus.
+
+    Public models are queryable through Campus API endpoints,
+    and may be returned by the Python API.
+    """
     id: schema.CampusID | schema.UserID
     created_at: schema.DateTime = dataclasses.field(
         default_factory=schema.DateTime.utcnow

campus/storage/tables/backend/postgres.py

@@ -31,7 +31,7 @@
 
 from campus.common import devops, env
 from campus.common.utils import datacls
-from campus.model import Model, constraints
+from campus.model import InternalModel, Model, constraints
 from campus.storage import errors
 
 from ..interface import PK, TableInterface
@@ -70,7 +70,8 @@ def _field_to_sql_schema(field: dataclasses.Field) -> str:
     constraints_sql = " ".join(sql_field_constraints)
     return f"\"{field_name}\" {sql_type} {constraints_sql}"
 
-def _model_to_sql_schema(name: str, model: type[Model]) -> str:
+
+def _model_to_sql_schema(name: str, model: type[InternalModel | Model]) -> str:
     """Convert a dataclass model to SQL schema."""
     columns = []
     constraints_ = []
@@ -286,7 +287,11 @@ def delete_matching(self, query: dict) -> None:
                     conn.commit()
 
     @devops.block_env(devops.PRODUCTION)
-    def init_from_model(self, name: str, model: type[Model]) -> None:
+    def init_from_model(
+            self,
+            name: str,
+            model: type[InternalModel | Model]
+    ) -> None:
         """Initialize the table from a Campus model definition."""
         create_table_sql = _model_to_sql_schema(name, model)
         # Ensure connection is properly closed after operation

campus/storage/tables/backend/sqlite.py

@@ -30,7 +30,7 @@
 
 from campus.common import devops
 from campus.common.utils import datacls
-from campus.model import Model, constraints
+from campus.model import InternalModel, Model, constraints
 from ..interface import TableInterface, PK
 
 
@@ -97,7 +97,7 @@ def _field_to_sql_schema(field: dataclasses.Field) -> str:
     return f"\"{field_name}\" {sql_type} {constraints_sql}"
 
 
-def _model_to_sql_schema(name: str, model: type[Model]) -> str:
+def _model_to_sql_schema(name: str, model: type[InternalModel | Model]) -> str:
     """Convert a dataclass model to SQL schema."""
     columns = []
     constraints_ = []
@@ -312,7 +312,7 @@ def delete_matching(self, query: Dict[str, Any]):
             self.delete_by_id(row[PK])
 
     @devops.block_env(devops.PRODUCTION)
-    def init_from_model(self, name: str, model: type[Model]) -> None:
+    def init_from_model(self, name: str, model: type[InternalModel | Model]) -> None:
         """Initialize the table from a Campus model definition."""
         conn = self.get_connection()
         create_table_sql = _model_to_sql_schema(name, model)

campus/storage/tables/interface.py

@@ -9,7 +9,7 @@
 
 from abc import ABC, abstractmethod
 
-from campus.model import Model
+from campus.model import InternalModel, Model
 
 # This constant should match the one in campus.common.schema
 PK = "id"
@@ -58,6 +58,6 @@ def delete_matching(self, query: dict):
         ...
 
     @abstractmethod
-    def init_from_model(self, name: str, model: type[Model]) -> None:
+    def init_from_model(self, name: str, model: type[InternalModel | Model]) -> None:
         """Initialize the table from a Campus model definition."""
         ...