diff --git a/campus/README.md b/campus/README.md index 8b5e27f1..40ad1c5e 100644 --- a/campus/README.md +++ b/campus/README.md @@ -20,16 +20,18 @@ campus/ ├── audit/ # Audit/tracing service + web UI ├── flask_campus/ # Flask integration glue ├── integrations/ # External service configurations -├── model/ # Entity representation (dataclasses) ├── services/ # Email, etc. ├── storage/ # Data persistence ├── webauth/ # WebAuth request primitives (headers, OAuth tokens) └── yapper/ # Event bus (self-contained) ``` -The shared kernel lives in its own distribution: +Two layers live in their own distributions, consumed as version-pinned +release wheels: [campus-common](https://github.com/nyjc-computing/campus-common) ships -`campus.common` and is consumed as a version-pinned release wheel. +the `campus.common` shared kernel, and +[campus-model](https://github.com/nyjc-computing/campus-model) ships +the `campus.model` entity dataclasses. ### Design Principles @@ -45,7 +47,7 @@ The shared kernel lives in its own distribution: - **campus.auth**: Authentication, OAuth, credentials (`.resources` for business logic) - **campus.api**: Circles, email OTP (`.resources` for business logic) - **campus.storage**: Backend-agnostic data persistence -- **campus.model**: Entity dataclasses (no business logic) +- **campus.model**: Entity dataclasses (no business logic) — [campus-model](https://github.com/nyjc-computing/campus-model) distribution - **campus.services**: Email, notifications - **campus.yapper**: Event bus — self-contained, imports no campus package ([docs/yapper.md](../docs/yapper.md)) @@ -72,7 +74,7 @@ The shared kernel lives in its own distribution: ### 🏛️ Model (`campus.model`) -Entity dataclasses: +Entity dataclasses (shipped by the [campus-model](https://github.com/nyjc-computing/campus-model) distribution): - User, Circle, Client, Session, Token - EmailOTP, HTTP headers, credentials - No business logic diff --git a/campus/model/__init__.py b/campus/model/__init__.py deleted file mode 100644 index fd0f0dae..00000000 --- a/campus/model/__init__.py +++ /dev/null @@ -1,90 +0,0 @@ -"""campus.model - -Models represent single entities in Campus API schema. - -They are typically represented as a row in a table, or a document in a -document store. -More complex models may span multiple tables or collections. - -Models are represented with dataclasses. -init parameters must be keyword-only; order of parameters should not -matter. -Models are expected to be used across the codebase; they should have -minimal dependencies, ideally none. Data processing logic should be -kept out of models. - -Exception: the HTTP header representations (campus.model.http) are -dict/str-based value classes rather than dataclasses. -""" - -__all__ = [ - "APIKey", - "AccessGrant", - "AppCredentials", - "Assignment", - "AuthSession", - "Circle", - "ClassroomLink", - "Client", - "ClientAccess", - "DeviceCode", - "EmailOTP", - "Feedback", - "HttpAuthProperty", - "HttpHeader", - "HttpHeaderWithAuth", - "Integration", - "InternalModel", - "LessonGroup", - "LessonGroupMember", - "LoginSession", - "Model", - "OAuthToken", - "Question", - "RateLimitBucket", - "Response", - "Submission", - "Timetable", - "TimetableEntry", - "TimetableMetadata", - "TraceSpan", - "TraceSummary", - "TraceTree", - "TraceTreeNode", - "User", - "UserCredentials", - "Vault", - "Venue", - "VenueBooking", -] - -from .audit import APIKey, RateLimitBucket, TraceSpan, TraceSummary, TraceTree, TraceTreeNode -from .base import InternalModel, Model -from .booking import Venue, VenueBooking -from .circle import Circle -from .classroom import ( - Assignment, - ClassroomLink, - Feedback, - Question, - Response, - Submission, -) -from .client import Client, ClientAccess -from .credentials import AppCredentials, OAuthToken, UserCredentials -from .device_code import DeviceCode -from .emailotp import EmailOTP -from .grant import AccessGrant -from .http import HttpAuthProperty, HttpHeader, HttpHeaderWithAuth -from .integration import Integration -from .login import LoginSession -from .session import AuthSession -from .timetable import ( - LessonGroup, - LessonGroupMember, - Timetable, - TimetableEntry, - TimetableMetadata, -) -from .user import User -from .vault import Vault diff --git a/campus/model/audit.py b/campus/model/audit.py deleted file mode 100644 index 5a6aaa1c..00000000 --- a/campus/model/audit.py +++ /dev/null @@ -1,640 +0,0 @@ -"""campus.model.audit - -Audit-related models for Campus API. - -These models represent trace spans and computed views/aggregations -for the audit service, including trace trees and summaries. -""" - -import datetime -import typing -from dataclasses import dataclass, field - -from campus.common import schema -from campus.common.utils import uid, utc_time -from campus.model.base import InternalModel, Model - -__all__ = [ - "APIKey", - "RateLimitBucket", - "TraceSpan", - "TraceTreeNode", - "TraceTree", - "TraceSummary", -] - -# Fallback for span records with an unparseable started_at: sorts before -# every real timestamp (all Campus timestamps are tz-aware UTC), so such -# spans become the tree root rather than disappearing. -_EPOCH_FALLBACK = datetime.datetime.min.replace(tzinfo=datetime.timezone.utc) - - -def _parse_started_at(span: dict) -> datetime.datetime | None: - """Parse a span record's started_at into an aware UTC datetime. - - Storage records carry the ISO string; in-memory dicts may already - hold a schema.DateTime (a str subclass). - """ - value = span.get("started_at") - if isinstance(value, schema.DateTime): - return value.to_datetime() - try: - return schema.DateTime(str(value)).to_datetime() - except (TypeError, ValueError): - return None - - -def _span_start_key(span: dict) -> tuple[datetime.datetime, str]: - """Deterministic sort key for spans: (started_at, span_id).""" - parsed = _parse_started_at(span) - return ( - parsed if parsed is not None else _EPOCH_FALLBACK, - str(span.get("span_id") or ""), - ) - - -@dataclass(eq=False, kw_only=True) -class RateLimitBucket(InternalModel): - """Token-bucket state for one audit rate-limit key (#538 Phase 3, #831). - - One row per active identity bucket; consumed by - campus.audit.resources.ratelimit before span ingestion. - - Attributes: - id: Encoded bucket key (storage PK). Encodes the identity tuple - ("client"/"user") or the producer API key fallback. - tokens: Remaining burst capacity, in spans. - last_refill: Unix epoch seconds of the last refill. - """ - - id: str = field( - default="", - metadata={"resource": False} # PK only, hidden from API - ) - tokens: float = 0.0 - last_refill: float = 0.0 - - -@dataclass(eq=False, kw_only=True) -class APIKey(Model): - """An API key issued for the campus.audit service. - - Only the key hash is stored; the plaintext value is shown once at - creation and never persisted. Lifecycle is tracked through - expires_at/revoked_at, and access is scoped via the scopes list - (e.g. "traces:read", "traces:*"). - - Issue: #566 - """ - id: schema.CampusID = field(default_factory=( - lambda: uid.generate_category_uid("apikey", length=16) - )) - # created_at: schema.DateTime is inherited from Model - # key_hash is stored in place of plaintext key for security - # and is not exposed via API resources - key_hash: schema.String = field(metadata={"resource": False}) - name: schema.String - owner_id: schema.UserID - scopes: list[schema.String] = field(default_factory=list) - # expires_at, revoked_at, and last_used are immutable audit fields - # They can be set through internal resource but not via the API. - expires_at: schema.DateTime | None = field( - default=None, - metadata={"mutable": False} - ) - revoked_at: schema.DateTime | None = field( - default=None, - metadata={"mutable": False} - ) - last_used: schema.DateTime | None = field( - default=None, - metadata={"mutable": False} - ) - # Requests per minute, None for unlimited - rate_limit: schema.Integer | None = None - - def is_active(self) -> bool: - """Check if the API key is currently active (not expired or - revoked). - - Returns: - bool: True if the API key is active, False otherwise. - """ - return not self.is_expired() and not self.is_revoked() - - def is_expired(self) -> bool: - """Check if the API key is expired. - - A key with no expires_at never expires. - - Returns: - bool: True if the API key is expired, False otherwise. - """ - if self.expires_at is None: - return False - # Storage records carry plain strings for union-annotated - # fields (from_storage only coerces single-class annotations) - expires_at = self.expires_at - if isinstance(expires_at, str): - expires_at = schema.DateTime(expires_at) - return expires_at.to_datetime() < utc_time.now() - - def is_revoked(self) -> bool: - """Check if the API key is revoked. - - Returns: - bool: True if the API key is revoked, False otherwise. - """ - return self.revoked_at is not None - - def has_scope(self, scope: str) -> bool: - """Check if the API key has a specific scope. - - Matching is exact: wildcard scopes stored on the key (e.g. - "traces:*", "*") only match themselves. Wildcard expansion is - the authorization layer's responsibility (#575). - - Args: - scope (str): The scope to check for. - - Returns: - bool: True if the API key has the scope, False otherwise. - - Raises: - TypeError: If the scope is not a string. - """ - if not isinstance(scope, str): - raise TypeError( - f"Scope must be a string, got {type(scope).__name__}" - ) - return scope in self.scopes - - @classmethod - def validate_update(cls, update: dict[str, typing.Any]) -> None: - """Validate an update dictionary against the APIKey model. - - Extends the base mutability check with type validation for the - mutable fields. - - Args: - update: Dictionary of fields to update - - Raises: - ValueError: If any field is not mutable or has an invalid - value. - """ - super().validate_update(update) - if "scopes" in update: - scopes = update["scopes"] - if not isinstance(scopes, list) or not all( - isinstance(s, str) for s in scopes - ): - raise ValueError( - "scopes must be a list of scope strings" - ) - if "rate_limit" in update: - rate_limit = update["rate_limit"] - if rate_limit is not None and ( - isinstance(rate_limit, bool) or not isinstance(rate_limit, int) - ): - raise ValueError( - "rate_limit must be an integer or None" - ) - - -@dataclass(eq=False, kw_only=True) -class TraceSpan(InternalModel): - """A single span in a distributed trace. - - Captures HTTP request-response data for observability and debugging. - Uses OpenTelemetry-compatible trace/span ID formats. - - Note: The `id` field is a Campus storage convention that mirrors `span_id`. - The `span_id` field follows OpenTelemetry conventions (16-char hex). - - Attributes: - id: Alias for span_id (Campus storage convention, stored as 'id' column) - trace_id: 32-char hex string grouping related spans - span_id: 16-char hex string identifying this span (OpenTelemetry convention) - parent_span_id: 16-char hex string of parent span, null for roots - method: HTTP method (GET, POST, etc.) - path: Request path (e.g. /api/v1/students) - query_params: Query string parameters as dict - request_headers: Request headers (Authorization stripped) - request_body: Request body JSON (if applicable) - status_code: HTTP response status code - response_headers: Response headers - response_body: Response body JSON (truncated to 64KB) - started_at: Span start timestamp - duration_ms: Span duration in milliseconds - api_key_id: API key used for the request - client_id: OAuth client identifier (null until auth succeeds) - user_id: User identifier (null until auth succeeds) - client_ip: Client IP address - user_agent: User-Agent header value - error_message: Error details for failed requests - tags: Arbitrary key-value metadata - """ - - # Trace identification (OpenTelemetry-compatible) - id: str = field( - default="", - metadata={"resource": False} # Hide from API, use as PK only - ) - trace_id: str = field(default_factory=uid.generate_trace_id) # 32-char hex - # 16-char hex (OpenTelemetry identifier, used in resources) - span_id: str = field(default_factory=uid.generate_span_id) - parent_span_id: str | None = None - - # Request data - method: str # HTTP method - path: str # Request path - query_params: dict[str, object] = field(default_factory=dict) - request_headers: dict[str, str] = field(default_factory=dict) - request_body: dict[str, object] | None = None - - # Response data - status_code: int | None = None - response_headers: dict[str, str] = field(default_factory=dict) - response_body: dict[str, object] | None = None - - # Timing - started_at: schema.DateTime = field(default_factory=schema.DateTime.utcnow) - duration_ms: float # milliseconds - - # Identity (nullable - populated after auth) - api_key_id: str | None = None - client_id: str | None = None - user_id: str | None = None - - # Client info - client_ip: str # INET type - user_agent: str | None = None - - # Error info - error_message: str | None = None - - # Metadata - tags: dict[str, object] = field(default_factory=dict) - - def __post_init__(self): - """Alias id to span_id after initialization. - - This ensures the Campus storage convention (id field as PK) - aligns with OpenTelemetry convention (span_id as identifier). - """ - if self.id == "" or self.id is None: - # Use object.__setattr__ to avoid frozen dataclass issues - object.__setattr__(self, "id", self.span_id) - - -@dataclass(eq=False, kw_only=True) -class TraceTreeNode(InternalModel): - """A node in a trace tree hierarchy. - - Represents a span with its children and tree metrics. - Computed from storage, not stored directly. - - Attributes: - span_id: 16-char hex span identifier - trace_id: 32-char hex trace identifier - parent_span_id: Parent span ID, None for root - method: HTTP method - path: Request path - status_code: HTTP response status - started_at: Span start timestamp - duration_ms: Span duration in milliseconds - error_message: Error details if applicable - children: Nested child spans - depth: Depth in tree (0 for root) - offset: Offset from parent start for visualization - """ - - # Span data (from TraceSpan) - span_id: str - trace_id: str - parent_span_id: str | None - method: str - path: str - status_code: int | None - started_at: schema.DateTime - duration_ms: float - error_message: str | None - - # Identity (#823): the summary path exposes these via the full - # TraceSpan; the tree path must too, or the detail page and the - # versioned detail API disagree with the list on the same spans. - client_id: str | None = None - user_id: str | None = None - # Client info (#826): same rationale — the detail metadata shows - # the requesting user agent, which only the full span carried. - user_agent: str | None = None - # Device (#825): same rationale — the detail metadata shows the - # device the request is attributed to. Lives in span tags, so the - # tree builder lifts it out to a first-class field. - device_id: str | None = None - - # Tree structure - children: "list[TraceTreeNode]" = field(default_factory=list) - - # Tree metrics (computed) - depth: int = 0 - offset: float = 0.0 - - def to_resource(self) -> dict[str, typing.Any]: - """Convert to resource dictionary for JSON response. - - Returns: - Dictionary with nested children arrays - """ - return { - "span_id": self.span_id, - "trace_id": self.trace_id, - "parent_span_id": self.parent_span_id, - "method": self.method, - "path": self.path, - "status_code": self.status_code, - "started_at": self.started_at, # DateTime is already a string - "duration_ms": self.duration_ms, - "error_message": self.error_message, - "client_id": self.client_id, - "user_id": self.user_id, - "user_agent": self.user_agent, - "children": [child.to_resource() for child in self.children], - "depth": self.depth, - "offset": self.offset, - } - - @classmethod - def from_storage(cls: type[typing.Self], record: dict[str, typing.Any]) -> typing.Self: - """TraceTreeNode is computed, not stored. Use TraceTree.from_spans() instead.""" - _ = record # Unused - computed view, not stored - raise NotImplementedError( - "TraceTreeNode is computed from spans, not loaded from storage. " - "Use TraceTree.from_spans() to build trace trees." - ) - - def to_storage(self) -> dict[str, typing.Any]: - """TraceTreeNode is computed, not stored.""" - raise NotImplementedError( - "TraceTreeNode is a computed view and cannot be stored." - ) - - @classmethod - def from_resource(cls: type[typing.Self], resource: dict[str, typing.Any]) -> typing.Self: - """TraceTreeNode is computed from spans, not created from resources.""" - _ = resource # Unused - computed view, not from resources - raise NotImplementedError( - "TraceTreeNode is computed from spans, not created from API resources. " - "Use TraceTree.from_spans() to build trace trees." - ) - - -@dataclass(eq=False, kw_only=True) -class TraceTree(InternalModel): - """A hierarchical trace tree with computed metrics. - - Represents a complete trace with all spans organized - in a tree structure with depth and offset metrics. - - Attributes: - root: Root span node with nested children, or None if empty - """ - - root: TraceTreeNode | None - - def to_resource(self) -> dict[str, typing.Any]: - """Convert to resource dictionary for JSON response. - - Returns: - Root span dict with nested children, or empty dict if no root - """ - if self.root is None: - return {} - return self.root.to_resource() - - @classmethod - def from_spans(cls: type[typing.Self], spans: list[dict]) -> typing.Self: - """Build a trace tree from flat span list. - - The root is the earliest-started span without a parent. Spans - whose parent was not ingested (or additional root spans, which - can appear in traces assembled from propagated X-Request-IDs) - are adopted under the root, ordered by started_at, so every - span stays visible in the waterfall (#794). - - Args: - spans: Flat list of span records from storage - - Returns: - TraceTree instance with root or None - """ - if not spans: - return cls(root=None) - - # Build span map for O(1) lookup - span_map: dict[str, dict] = {s["span_id"]: {**s, "children": []} for s in spans} - - # Wire children to parents; collect spans with no parent in the - # set (root spans, or orphans whose parent is missing) as tops. - tops: list[dict] = [] - for span in spans: - node = span_map[span["span_id"]] - parent_id = span.get("parent_span_id") - if parent_id is None or parent_id == span["span_id"]: - # Root span (normalize self-parenting, which would cycle) - node["parent_span_id"] = None - tops.append(node) - elif parent_id in span_map: - span_map[parent_id]["children"].append(node) - else: - # Orphan: parent span not ingested; adopt under the root - node["parent_span_id"] = None - tops.append(node) - - if not tops: - # Defensive: every span is somebody's child, which requires a - # parent cycle. Break it at the earliest span. - earliest = min(span_map.values(), key=_span_start_key) - parent_id = earliest.get("parent_span_id") - earliest["parent_span_id"] = None - if parent_id in span_map: - siblings = span_map[parent_id]["children"] - siblings[:] = [ - s for s in siblings if s["span_id"] != earliest["span_id"] - ] - tops = [earliest] - - tops.sort(key=_span_start_key) - root_dict = tops[0] - root_dict["children"].extend(tops[1:]) - root_dict["children"].sort(key=_span_start_key) - root = cls._build_node(root_dict, 0, 0.0) - - return cls(root=root) - - @classmethod - def _build_node( - cls, - span_dict: dict, - depth: int, - offset: float, - ) -> TraceTreeNode: - """Build a TraceTreeNode from span dict with children. - - Args: - span_dict: Span dict with children array - depth: Current depth in tree - offset: Offset from parent start in ms - - Returns: - TraceTreeNode with nested children - """ - parent_start = _parse_started_at(span_dict) - child_nodes: list[TraceTreeNode] = [] - - for child_dict in span_dict.get("children", []): - child_start = _parse_started_at(child_dict) - if parent_start is not None and child_start is not None: - child_offset = max(0.0, (child_start - parent_start).total_seconds() * 1000) - else: - child_offset = 0.0 - child = cls._build_node(child_dict, depth + 1, child_offset) - child_nodes.append(child) - - node = TraceTreeNode( - span_id=span_dict["span_id"], - trace_id=span_dict["trace_id"], - parent_span_id=span_dict.get("parent_span_id"), - method=span_dict["method"], - path=span_dict["path"], - status_code=span_dict.get("status_code"), - started_at=span_dict["started_at"], - duration_ms=span_dict.get("duration_ms", 0), - error_message=span_dict.get("error_message"), - client_id=span_dict.get("client_id"), - user_id=span_dict.get("user_id"), - user_agent=span_dict.get("user_agent"), - device_id=(span_dict.get("tags") or {}).get("device_id"), - children=child_nodes, - depth=depth, - offset=offset, - ) - - return node - - @classmethod - def from_storage(cls: type[typing.Self], record: dict[str, typing.Any]) -> typing.Self: - """TraceTree is computed, not stored. Use TraceTree.from_spans() instead.""" - _ = record # Unused - computed view, not stored - raise NotImplementedError( - "TraceTree is computed from spans, not loaded from storage. " - "Use TraceTree.from_spans(spans) to build trace trees." - ) - - def to_storage(self) -> dict[str, typing.Any]: - """TraceTree is computed, not stored.""" - raise NotImplementedError( - "TraceTree is a computed view and cannot be stored." - ) - - @classmethod - def from_resource(cls: type[typing.Self], resource: dict[str, typing.Any]) -> typing.Self: - """TraceTree is computed from spans, not created from resources.""" - _ = resource # Unused - computed view, not from resources - raise NotImplementedError( - "TraceTree is computed from spans, not created from API resources. " - "Use TraceTree.from_spans(spans) to build trace trees." - ) - - -@dataclass(eq=False, kw_only=True) -class TraceSummary(InternalModel): - """A summary of a trace for listing/search results. - - Computed from spans in a trace, includes root span - and aggregate metrics. - - Attributes: - trace_id: 32-char hex trace identifier - span_count: Total number of spans in trace - started_at: Root span start timestamp - duration_ms: Root span duration - root_span: The root span details - """ - - trace_id: str - span_count: int - started_at: schema.DateTime - duration_ms: float - root_span: TraceSpan - - def to_resource(self) -> dict[str, typing.Any]: - """Convert to resource dictionary for JSON response.""" - return { - "trace_id": self.trace_id, - "span_count": self.span_count, - "started_at": self.started_at, # DateTime is already a string - "duration_ms": self.duration_ms, - "root_span": self.root_span.to_resource(), - } - - @classmethod - def from_spans(cls: type[typing.Self], trace_id: str, spans: list[dict]) -> typing.Self: - """Build a summary from spans grouped by trace_id. - - Args: - trace_id: The trace identifier - spans: List of span records for this trace - - Returns: - TraceSummary instance - """ - # Find root span (parent_span_id is None) - root_span_dict = next( - (s for s in spans if s.get("parent_span_id") is None), - None - ) - - if root_span_dict: - root_span = TraceSpan.from_storage(root_span_dict) - duration_ms = root_span_dict["duration_ms"] - started_at = root_span_dict["started_at"] - else: - # No root span, use earliest span - root_span_dict = min(spans, key=lambda s: s["started_at"]) - root_span = TraceSpan.from_storage(root_span_dict) - duration_ms = sum(s.get("duration_ms", 0) for s in spans) - started_at = root_span_dict["started_at"] - - return cls( - trace_id=trace_id, - span_count=len(spans), - started_at=started_at, - duration_ms=duration_ms, - root_span=root_span, - ) - - @classmethod - def from_storage(cls: type[typing.Self], record: dict[str, typing.Any]) -> typing.Self: - """TraceSummary is computed, not stored. Use TraceSummary.from_spans() instead.""" - _ = record # Unused - computed view, not stored - raise NotImplementedError( - "TraceSummary is computed from spans, not loaded from storage. " - "Use TraceSummary.from_spans(trace_id, spans) to build summaries." - ) - - def to_storage(self) -> dict[str, typing.Any]: - """TraceSummary is computed, not stored.""" - raise NotImplementedError( - "TraceSummary is a computed view and cannot be stored." - ) - - @classmethod - def from_resource(cls: type[typing.Self], resource: dict[str, typing.Any]) -> typing.Self: - """TraceSummary is computed from spans, not created from resources.""" - _ = resource # Unused - computed view, not from resources - raise NotImplementedError( - "TraceSummary is computed from spans, not created from API resources. " - "Use TraceSummary.from_spans(trace_id, spans) to build summaries." - ) diff --git a/campus/model/base.py b/campus/model/base.py deleted file mode 100644 index afb6ff25..00000000 --- a/campus/model/base.py +++ /dev/null @@ -1,221 +0,0 @@ -"""campus.model.base - -Base Model class. -""" - -import dataclasses -import types -import typing -from dataclasses import dataclass - -from campus.common import schema - - -def _coercible_str_type(hint: typing.Any) -> type | None: - """Resolve the str subclass a type hint declares, if any. - - Handles plain annotations (schema.DateTime) and Optional unions - (schema.DateTime | None): storage/resource round-trips keep no - subclass, so Optional[schema.*] fields must coerce too (#847). - Returns None for non-str hints (Model, list, bare unions of - multiple non-None members, ...). - """ - field_type = hint - origin = typing.get_origin(field_type) - # PEP 604 `X | None` unions have origin types.UnionType; subscripted - # typing.Union has origin typing.Union. Both are Optional-style. - if origin is typing.Union or origin is types.UnionType: - members = [ - arg for arg in typing.get_args(field_type) - if arg is not type(None) - ] - if len(members) != 1: - return None - field_type = members[0] - if isinstance(field_type, type) and issubclass(field_type, str): - return field_type - return None - - -class FieldMeta(typing.TypedDict): - """Metadata for a model field. - - These are passed to the metadata parameter of dataclasses.field(). - See https://docs.python.org/3/library/dataclasses.html#dataclasses.field - - Note: Types are enforced when generating SQL schemas via - TableInterface.init_from_model(). Invalid metadata will raise - TypeError or ValueError. - - Attributes: - resource: Whether the field is returned in API responses. Default is True. - storage: Whether the field is stored in the database. Must be bool. Default is True. - constraints: List of constraint names. Must be list/tuple of str. - Valid constraint names are defined in campus.model.constraints - (e.g., "unique" for UNIQUE constraint). - """ - # Whether the field is returned in API responses. Default is True. - resource: bool - # Whether the field is stored in the database. Must be bool. Default is True. - storage: bool - # Any additional constraints for the field. Must be list/tuple of str. - # Valid values are defined in campus.model.constraints (e.g., "unique"). - 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]: - return {field.name: field for field in dataclasses.fields(cls)} - - @classmethod - def validate_update(cls, update: dict[str, typing.Any]) -> None: - """Validate an update dictionary against the model's mutable - fields. - - Args: - update: Dictionary of fields to update - - Raises: - ValueError: If any field in the update is not mutable - """ - if not update: - raise ValueError("No fields provided for update validation") - for field_name in update: - field = cls.fields().get(field_name) - if field is None: - raise ValueError( - f"Field '{field_name}' does not exist in model" - ) - if not field.metadata.get("mutable", True): - raise ValueError(f"Field '{field_name}' is not mutable") - - @classmethod - def from_resource( - cls: type[typing.Self], - resource: dict[str, typing.Any] - ) -> typing.Self: - """Create a model instance from a resource dictionary. - - Fields declared init=False are assigned after construction, - since they cannot be passed to __init__(). Values for - Model-typed fields are deserialized recursively, and plain - strings are coerced to annotated str subclasses - (e.g. schema.DateTime). - """ - try: - hints = typing.get_type_hints(cls) - except Exception: - # Unresolvable type hints disable coercion below - hints = {} - - init_kwargs: dict[str, typing.Any] = {} - post_init_fields: dict[str, typing.Any] = {} - for field in cls.fields().values(): - if not field.metadata.get("resource", True): - continue - if field.name not in resource: - continue - value = resource[field.name] - field_type = hints.get(field.name) - if (isinstance(value, dict) and isinstance(field_type, type) - and issubclass(field_type, Model)): - value = field_type.from_resource(value) - else: - str_type = _coercible_str_type(field_type) - if str_type is not None and isinstance(value, str) \ - and not isinstance(value, str_type): - value = str_type(value) - if field.init: - init_kwargs[field.name] = value - else: - post_init_fields[field.name] = value - - instance = cls(**init_kwargs) - for field_name, value in post_init_fields.items(): - setattr(instance, field_name, value) - return instance - - @classmethod - def from_storage( - cls: type[typing.Self], - record: dict[str, typing.Any] - ) -> typing.Self: - """Create a model instance from a storage record dictionary. - - Storage records carry plain strings (JSON/BSON); values for - fields annotated as str subclasses (schema.DateTime, CampusID, - Email, Url, ...) are coerced to the annotated type, mirroring - from_resource(). Optional annotations (schema.DateTime | None) - coerce too (#847) — storage round-trips keep no subclass. - """ - try: - hints = typing.get_type_hints(cls) - except Exception: - # Unresolvable type hints disable coercion below - hints = {} - - def get_value(f: dataclasses.Field) -> typing.Any: - """Get value from record, falling back to field default if missing.""" - if f.name in record: - value = record[f.name] - # Key not in record - use field default if available - elif f.default is not dataclasses.MISSING: - return f.default - elif f.default_factory is not dataclasses.MISSING: # type: ignore[attr-defined] - return f.default_factory() # type: ignore[attr-defined] - # No default available - raise KeyError with clear message - else: - raise KeyError( - f"Required field '{f.name}' not found in storage record " - f"for model '{cls.__name__}'" - ) - str_type = _coercible_str_type(hints.get(f.name)) - if str_type is not None and isinstance(value, str) \ - and not isinstance(value, str_type): - value = str_type(value) - return value - - return cls( - **{ - field.name: get_value(field) - for field in cls.fields().values() - if field.metadata.get("storage", True) - } - ) - - def to_resource(self) -> dict[str, typing.Any]: - """Convert the model instance to a resource dictionary.""" - return { - field.name: getattr(self, field.name) - for field in self.fields().values() - if field.metadata.get("resource", True) - } - - def to_storage(self) -> dict[str, typing.Any]: - """Convert the model instance to a storage record dictionary.""" - return { - field.name: getattr(self, field.name) - for field in self.fields().values() - if field.metadata.get("storage", True) - } - - -@dataclass(kw_only=True) -class Model(InternalModel): - """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 - ) diff --git a/campus/model/booking.py b/campus/model/booking.py deleted file mode 100644 index e6ec9d73..00000000 --- a/campus/model/booking.py +++ /dev/null @@ -1,64 +0,0 @@ -"""campus.model.booking - -Venue and venue-booking entity models. -""" - -from dataclasses import dataclass, field - -from campus.common import schema -from campus.common.utils import uid - -from . import constraints -from .base import Model - - -@dataclass(eq=False, kw_only=True) -class VenueBooking(Model): - """ - Represents a booking for a venue on a specific date and time. - - Fields: - id (CampusID): Unique identifier for the booking. - user_id (UserID): FK referencing the user who made the booking. - venue_id (CampusID): FK referencing the venue being booked. - description (String): Description or purpose of the booking. - start_time (Time): Start time of the booking. - end_time (Time): End time of the booking. - date (Date): Date of the booking. - - Constraints: - - Unique by venue_id, date, start_time, and end_time to prevent exact duplicate bookings. - Overlap checking must be conducted separately. - """ - id: schema.CampusID = field(default_factory=( - lambda: uid.generate_category_uid("booking", length=8) - )) - user_id: schema.UserID - venue_id: schema.CampusID - description: schema.String - start_time: schema.Time - end_time: schema.Time - date: schema.Date - # NOTE: This only prevents exact duplicate bookings. - # Overlap checking must still be conducted separately - __constraints__ = constraints.Unique("venue_id", "date", "start_time", "end_time") - - -@dataclass(eq=False, kw_only=True) -class Venue(Model): - """ - Represents a venue that can be booked. - - Fields: - id (CampusID): Unique identifier for the venue. - venue_name (String): Name of the venue. - - Note: - Other model fields to be confirmed (TBC). - """ - id: schema.CampusID = field(default_factory=( - lambda: uid.generate_category_uid("venue", length=8) - )) - venue_name: schema.String - venue_type: schema.String - diff --git a/campus/model/circle.py b/campus/model/circle.py deleted file mode 100644 index 191bd86f..00000000 --- a/campus/model/circle.py +++ /dev/null @@ -1,27 +0,0 @@ -"""campus.model.circle - -Circle model for the Campus API. - -This module defines circles (organizational units/groups) for the Campus API. -""" - -from dataclasses import dataclass, field - -from campus.common import schema -from campus.common.utils import uid - -from .base import Model - - -@dataclass(eq=False, kw_only=True) -class Circle(Model): - """Dataclass representation of a circle record.""" - id: schema.CampusID = field(default_factory=( - lambda: uid.generate_category_uid("circle", length=8) - )) - # created_at inherited from Model - name: str - description: str = "" - tag: str # CircleTag type alias from models - members: dict[schema.CampusID, int] = field(default_factory=dict) - sources: dict = field(default_factory=dict) diff --git a/campus/model/classroom.py b/campus/model/classroom.py deleted file mode 100644 index d339ed85..00000000 --- a/campus/model/classroom.py +++ /dev/null @@ -1,399 +0,0 @@ -"""campus.model.classroom - -Classroom models for Campus Classroom (Google Classroom Add-On). - -This module contains all models related to the campus-classroom feature: -- Assignments with structured questions -- Student submissions with responses -- Teacher feedback -- Google Classroom integration links - -For schema details, see: -https://github.com/nyjc-computing/campus-classroom/blob/main/docs/schema-proposal.md -""" - -import re -from dataclasses import dataclass, field - -from campus.common import schema -from campus.common.utils import uid - -from .base import Model - -# ============================================================================ -# Assignment Models -# ============================================================================ - -@dataclass(eq=False, kw_only=True) -class Question: - """A single question within an assignment. - - Question IDs use hierarchical dot notation (q1, q1.a, q1.a.i). - Hierarchy is self-describing in the ID - no parent_id needed. - - Level 1: q1, q2, q3, ... (main questions) - Level 2: q1.a, q1.b, q1.c, ... (parts) - Level 3: q1.a.i, q1.a.ii, q1.b.i, ... (subparts) - Level 4+: q1.a.i.1, q1.a.i.2, ... (further nesting) - - type is a question-type id registered in campus.api.questionschemas - and data holds that type's configuration (e.g. multiple-choice - options). The model treats both as opaque: validity is enforced at - the API payload boundary, not here — reads construct models on every - fetch and must stay cheap and dependency-free. - """ - id: str # e.g., "q1", "q1.a", "q1.a.i" - prompt: str # Context/passage (may be empty) - question: str # The actual question/task - type: str = "free-response" # Registered question type id - data: dict | None = None # Type-specific configuration (JSON) - - def __post_init__(self) -> None: - """Validate question ID format.""" - # Validate hierarchical dot notation: alphanumeric with dots only - if not re.match(r'^[a-zA-Z0-9]+(\.[a-zA-Z0-9]+)*$', self.id): - raise ValueError( - f"Invalid question ID format: {self.id}. " - "Must use hierarchical dot notation (e.g., q1, q1.a, q1.a.i)" - ) - - @property - def level(self) -> int: - """Return the hierarchy level (1-indexed).""" - return len(self.id.split(".")) - - @property - def parent_id(self) -> str | None: - """Return the parent question ID, or None for top-level questions.""" - parts = self.id.split(".") - if len(parts) <= 1: - return None - return ".".join(parts[:-1]) - - @property - def root_id(self) -> str: - """Return the root question ID (e.g., 'q1' for 'q1.a.ii').""" - return self.id.split(".")[0] - - -@dataclass(eq=False, kw_only=True) -class ClassroomLink: - """Link to a Google Classroom assignment. - - One assignment can be linked to multiple Classroom classes. - Each link represents one CourseWork assignment in one course. - """ - course_id: str # Google Classroom course ID - coursework_id: str # Google Classroom assignment ID - attachment_id: str | None = None # Google Classroom attachment ID (if Add-On) - linked_at: schema.DateTime = field(default_factory=schema.DateTime.utcnow) - - -@dataclass(eq=False, kw_only=True) -class Assignment(Model): - """Dataclass representation of an assignment record. - - An assignment contains structured questions and can be linked - to one or more Google Classroom classes. - - Note: Deadlines are managed by Google Classroom, not stored here. - """ - id: schema.CampusID = field(default_factory=( - lambda: uid.generate_category_uid("assignment", length=8) - )) - # created_at inherited from Model - title: str - description: str = "" - questions: list[Question] = field(default_factory=list) - created_by: schema.UserID # Teacher who created the assignment - updated_at: schema.DateTime = field( - default_factory=schema.DateTime.utcnow, - metadata={"mutable": True} - ) - # Links to Google Classroom (one-to-many) - classroom_links: list[ClassroomLink] = field(default_factory=list) - - @classmethod - def from_resource(cls, resource: dict) -> "Assignment": - """Create an Assignment from an API resource response. - - Handles nested deserialization of questions and classroom_links. - """ - # Deserialize nested objects - questions = [ - Question(**q) if isinstance(q, dict) else q - for q in resource.get("questions", []) - ] - classroom_links = [ - ClassroomLink( - course_id=link["course_id"], - coursework_id=link["coursework_id"], - attachment_id=link.get("attachment_id"), - **({"linked_at": schema.DateTime(link["linked_at"])} if "linked_at" in link else {}) - ) if isinstance(link, dict) else link - for link in resource.get("classroom_links", []) - ] - - # Build kwargs for Assignment constructor - kwargs = { - "id": resource["id"], - "created_at": resource["created_at"], - "title": resource["title"], - "description": resource.get("description", ""), - "questions": questions, - "created_by": resource["created_by"], - "updated_at": resource["updated_at"], - "classroom_links": classroom_links, - } - return cls(**kwargs) - - @classmethod - def from_storage(cls, record: dict) -> "Assignment": - """Create an Assignment from a storage record. - - Storage records share the resource shape (nested questions and - classroom_links as dicts), so deserialization delegates to - from_resource(). - """ - return cls.from_resource(record) - - def to_resource(self) -> dict: - """Convert the Assignment to an API resource response. - - Handles nested serialization of questions and classroom_links. - """ - from dataclasses import asdict - return { - "id": self.id, - "created_at": self.created_at, - "title": self.title, - "description": self.description, - "questions": [asdict(q) for q in self.questions], - "created_by": self.created_by, - "updated_at": self.updated_at, - "classroom_links": [asdict(link) for link in self.classroom_links], - } - - def to_storage(self) -> dict: - """Convert the Assignment to storage format. - - Handles serialization of nested dataclasses to dicts. - """ - from dataclasses import asdict - - data = super().to_storage() - # Convert nested dataclasses to dicts for storage - data['questions'] = [asdict(q) for q in self.questions] - data['classroom_links'] = [asdict(link) for link in self.classroom_links] - return data - - def get_question_tree(self) -> dict: - """Return questions as a nested tree structure.""" - # First pass: create all nodes - nodes = {} - for q in self.questions: - nodes[q.id] = { - "id": q.id, - "prompt": q.prompt, - "question": q.question, - "children": [] - } - - # Second pass: build tree structure - tree: dict = {} - for q in self.questions: - node = nodes[q.id] - if q.level == 1: - tree[q.id] = node - elif q.parent_id in nodes: - nodes[q.parent_id]["children"].append(node) - return tree - - def get_question(self, question_id: str) -> Question | None: - """Get a question by ID.""" - for q in self.questions: - if q.id == question_id: - return q - return None - - def get_questions_for_root(self, root_id: str) -> list[Question]: - """Get all questions under a root question (including subparts).""" - prefix = f"{root_id}." - return [ - q for q in self.questions - if q.id == root_id or q.id.startswith(prefix) - ] - - -# ============================================================================ -# Submission Models -# ============================================================================ - -@dataclass(eq=False, kw_only=True) -class Response: - """A student's response to a single question. - - References Assignment.question.id via question_id. - - response_data is the canonical answer payload, shaped by the - question's registered type ({"text": ...} for the text types, - {"choice":