Update to dataclasses & complete flask implementation

campus/api/resources/event.py

@@ -0,0 +1,140 @@
+"""campus.api.resources.event
+
+Event resource for Campus API.
+"""
+
+from campus.common import schema
+import campus.model
+import campus.storage
+
+event_storage = campus.storage.get_table("event")
+
+
+# class Event:
+#     """Event model for handling database operations related to events."""
+
+#     def __init__(self):
+#         """Initialize the User model with a table storage interface."""
+#         self.storage = get_table(TABLE)
+
+#     def new(
+#             self,
+#             *,
+#             name: str,
+#             location: str | None = None,
+#             location_url: str | None = None,
+#             start_time: schema.DateTime | None = None,
+#             duration: int | None = None,
+#             description: str,
+#     ) -> EventRecord:
+#         """Create a new event."""
+#         event_id = CampusID(uid.generate_category_uid("event", length=16))
+
+#         # Least ugly solution due to type gymnastics.
+#         dtnow = schema.DateTime.utcnow()
+#         record = EventRecord.from_dict({
+#             "id": event_id,
+#             "created_at": dtnow,
+#             "name": fields.name,
+#             "location": fields.location,
+#             "location_url": fields.location_url,
+#             "duration": fields.duration,
+#             "start_time": fields.start_time
+#         })
+
+#         try:
+#             self.storage.insert_one(record.to_dict())
+#             return record
+#         except storage_errors.ConflictError:
+#             raise api_errors.ConflictError(
+#                 message="Event with the same ID already exists",
+#                 event_id=event_id
+#             ) from None
+#         except Exception as e:
+#             raise api_errors.InternalError(message=str(e), error=e) from e
+
+#     def _try_get(self, event_id: CampusID) -> EventRecord:
+#         """Private method.
+#         Tries to get event by event_id
+#         Raises InternalError upon other errors."""
+#         try:
+#             event = self.storage.get_by_id(event_id)
+#             if not event:
+#                 raise api_errors.ConflictError(
+#                     message="Event not found",
+#                     event_id=event_id
+#                 )
+#             event = EventRecord(**event)  # Assert-coerce to EventResource.
+#         except api_errors.ConflictError:
+#             raise
+#         except Exception as e:
+#             raise api_errors.InternalError(message=str(e), error=e)
+
+#         return event
+
+#     def delete(self, id: CampusID, fields: EventDelete) -> None:
+#         """Delete an event by id."""
+#         self._try_get(id)  # Make sure it exists.
+#         try:
+#             self.storage.delete_by_id(id)
+#         except Exception as e:
+#             raise api_errors.InternalError(message=str(e), error=e)
+
+#     def get(self, id: CampusID, fields: EventGet) -> EventRecord:
+#         """Get an event by id."""
+#         return self._try_get(id)
+
+#     def update(self, id: CampusID, fields: EventUpdate) -> EventRecord:
+#         """Update an event by id."""
+#         # Check if user exists first
+
+#         self._try_get(id)  # Make sure it exists.
+#         try:
+#             self.storage.update_by_id(id, fields.to_dict())
+#             return self._try_get(id)
+#         except Exception as e:
+#             raise api_errors.InternalError(message=str(e), error=e)
+
+
+class EventsResource:
+    """Represents the events resource in Campus API Schema."""
+    # campus.events.new()
+
+    @staticmethod
+    def init_storage() -> None:
+        """Initialize storage for client authentication."""
+        event_storage.init_from_model("event", campus.model.Event)
+
+    def __getitem__(self, event_id: schema.CampusID) -> "EventResource":
+        """Get an Event resource by event ID."""
+        return EventResource()
+
+    def new(
+            self,
+            *,
+            name: str,
+            location: str | None = None,
+            location_url: str | None = None,
+            start_time: schema.DateTime | None = None,
+            duration: int | None = None,
+            description: str,
+    ) -> campus.model.Event:
+        """Create a new event and return the Event model instance."""
+        client = campus.model.Event(
+            name=name,
+            location=location,
+            location_url=location_url,
+            start_time=start_time,
+            duration=duration,
+            description=description,
+        )
+        event_storage.insert_one(client.to_storage())
+        return client
+
+
+
+class EventResource:
+    """Represents a single event resource in Campus API Schema."""
+    # campus.events[event_id].get()
+    # campus.events[event_id].update()
+    # campus.events[event_id].delete()

campus/apps/api/routes/events.py

@@ -0,0 +1,107 @@
+"""campus.apps.api.routes.events
+
+API routes for the events resource (abstract endpoints only).
+"""
+
+from typing import Callable
+
+from flask import Blueprint, Flask
+from flask import request as flask_request # For datetime wrapper
+
+import campus.common.validation.flask as flask_validation
+from campus.apps.campusauth import authenticate_client
+from campus.common.errors import api_errors
+from campus.models import event  # Event model to be implemented
+from campus.common.utils import utc_time
+import campus.yapper as campus_yapper # TODO: IMPORTANT: Add event_id context when yapper emits
+
+bp = Blueprint('events', __name__, url_prefix='/events')
+bp.before_request(authenticate_client)
+
+yapper = campus_yapper.create()
+events = event.Event()
+
+def init_app(app: Flask | Blueprint) -> None:
+    """Initialise event routes with the given Flask app/blueprint."""
+    app.register_blueprint(bp)
+
+@bp.post('/')
+def new_event(*_: str) -> flask_validation.JsonResponse:
+    """Create a new event."""
+
+    payload = flask_validation.validate_request_and_extract_json(
+        event.EventNew.__annotations__,
+        on_error=api_errors.raise_api_error,
+    )
+
+    record = events.new(event.EventNew.from_dict(payload))
+
+    flask_validation.validate_json_response(
+        event.EventRecord.__annotations__,
+        record.to_dict(),
+        on_error=api_errors.raise_api_error,
+    )
+
+    yapper.emit('campus.events.new')
+    return record.to_dict(), 200
+
+
+@bp.get('/<string:event_id>')
+def get_event_details(event_id: str) -> flask_validation.JsonResponse:
+    """Get details of an event occurrence."""
+
+    # Parse payload even though empty expected.
+    # To allow for future additions.
+    payload = flask_validation.validate_request_and_extract_json(
+        event.EventGet.__annotations__,
+        on_error=api_errors.raise_api_error,
+    )
+
+    record = events.get(event_id, event.EventGet.from_dict(payload))
+
+    flask_validation.validate_json_response(
+        event.EventRecord.__annotations__,
+        record.to_dict(),
+        on_error=api_errors.raise_api_error,
+    )
+
+    # No event emitted.
+    return record.to_dict(), 200
+
+
+# TODO: Refactor to not make all fields required (:cry:)
+@bp.patch('/<string:event_id>')
+def update_event(event_id: str) -> flask_validation.JsonResponse:
+    """Edit an event occurrence."""
+
+    payload = flask_validation.validate_request_and_extract_json(
+        event.EventUpdate.__annotations__,
+        on_error=api_errors.raise_api_error,
+    )
+
+    record = events.update(event_id, event.EventUpdate.from_dict(payload))
+
+    flask_validation.validate_json_response(
+        event.EventRecord.__annotations__,
+        record.to_dict(),
+        on_error=api_errors.raise_api_error,
+    )
+
+    yapper.emit('campus.events.update')
+    return record.to_dict(), 200
+
+
+@bp.delete('/<string:event_id>')
+def delete_event(event_id: str) -> flask_validation.JsonResponse:
+    """Delete an event occurrence."""
+    # Parse payload even though empty expected.
+    # To allow for future additions.
+    payload = flask_validation.validate_request_and_extract_json(
+        event.EventDelete.__annotations__,
+        on_error=api_errors.raise_api_error,
+    )
+
+    events.delete(event_id, event.EventDelete.from_dict(payload))
+
+    yapper.emit('campus.events.delete')
+    return {}, 204 # 204 right now because no content lol.

campus/common/utils/secret.py

@@ -154,7 +154,6 @@ def hash_client_secret(secret: str, key: str) -> str:
     hmac_hash = hmac.new(hmac_key, secret_bytes, hashlib.sha256).digest()
     return base64.urlsafe_b64encode(hmac_hash).decode('utf-8')
 
-
 def hash_otp(otp: str) -> str:
     """Hash the OTP using bcrypt for secure storage.
 
@@ -166,7 +165,6 @@ def hash_otp(otp: str) -> str:
     hashed = bcrypt.hashpw(otp_bytes, salt)
     return hashed.decode('utf-8')
 
-
 def verify_otp(plain_otp: str, hashed_otp: str) -> bool:
     """Verify if a plaintext OTP matches this hashed OTP.
 

campus/model/__init__.py

@@ -43,6 +43,7 @@
 from .credentials import OAuthToken, UserCredentials
 from .http.header import HttpHeader, HttpHeaderWithAuth
 from .emailotp import EmailOTP
+from .event import Event
 from .login import LoginSession
 from .session import AuthSession
 from .submission import Feedback, Response, Submission

campus/model/event.py

@@ -0,0 +1,38 @@
+"""campus.model.event
+
+Event model definitions for Campus.
+"""
+
+from typing import ClassVar
+from dataclasses import dataclass, field
+
+from campus.common import schema
+from campus.common.schema.openapi import String, DateTime
+from campus.common.utils import uid
+
+from .base import Model
+from . import constraints
+
+@dataclass(eq=False, kw_only=True)
+class Event(Model):
+    """
+    Represents a single instance of an event.
+    """
+    id: schema.CampusID = field(default_factory=(
+        lambda: uid.generate_category_uid("event", length=8)
+    ))
+
+    # Cosmetic fields
+    name: String
+    description: String
+    location: String
+
+    # Data fields
+    location_url: String | None = None # Optional, online events only
+    start_time: DateTime
+    end_time: DateTime
+
+    # No duplicate events allowed
+    __constraints__ = constraints.Unique(
+        "name", "description", "location", "location_url", "start_time", "end_time"
+    )