Extend Nest API with Snapshot data

Author: rudransh-shrivastava

backend/apps/api/rest/v0/__init__.py

@@ -16,6 +16,7 @@
 from apps.api.rest.v0.project import router as project_router
 from apps.api.rest.v0.release import router as release_router
 from apps.api.rest.v0.repository import router as repository_router
+from apps.api.rest.v0.snapshot import router as snapshot_router
 from apps.api.rest.v0.sponsor import router as sponsor_router
 
 ROUTERS = {
@@ -29,6 +30,7 @@
     "/projects": project_router,
     "/releases": release_router,
     "/repositories": repository_router,
+    "/snapshots": snapshot_router,
     "/sponsors": sponsor_router,
 }
 

backend/apps/api/rest/v0/snapshot.py

@@ -0,0 +1,226 @@
+"""Snapshot API."""
+
+from datetime import datetime
+from http import HTTPStatus
+from typing import Literal
+
+from django.http import HttpRequest
+from ninja import Path, Query, Schema
+from ninja.decorators import decorate_view
+from ninja.pagination import RouterPaginated
+from ninja.responses import Response
+
+from apps.api.decorators.cache import cache_response
+from apps.api.rest.v0.chapter import Chapter
+from apps.api.rest.v0.issue import Issue
+from apps.api.rest.v0.member import Member
+from apps.api.rest.v0.project import Project
+from apps.api.rest.v0.release import Release
+from apps.github.models.issue import Issue as IssueModel
+from apps.github.models.release import Release as ReleaseModel
+from apps.github.models.user import User as UserModel
+from apps.owasp.models.chapter import Chapter as ChapterModel
+from apps.owasp.models.project import Project as ProjectModel
+from apps.owasp.models.snapshot import Snapshot as SnapshotModel
+
+router = RouterPaginated(tags=["Snapshots"])
+
+
+class SnapshotBase(Schema):
+    """Base schema for Snapshot (used in list endpoints)."""
+
+    created_at: datetime
+    end_at: datetime
+    key: str
+    start_at: datetime
+    title: str
+    updated_at: datetime
+
+
+class Snapshot(SnapshotBase):
+    """Schema for Snapshot (minimal fields for list display)."""
+
+
+class SnapshotDetail(SnapshotBase):
+    """Detail schema for Snapshot (used in single item endpoints)."""
+
+    new_chapters_count: int
+    new_issues_count: int
+    new_projects_count: int
+    new_releases_count: int
+    new_users_count: int
+
+
+class SnapshotError(Schema):
+    """Snapshot error schema."""
+
+    message: str
+
+
+@router.get(
+    "/",
+    description="Retrieve a paginated list of OWASP snapshots.",
+    operation_id="list_snapshots",
+    response=list[Snapshot],
+    summary="List snapshots",
+)
+@decorate_view(cache_response())
+def list_snapshots(
+    request: HttpRequest,
+    ordering: Literal[
+        "created_at", "-created_at", "updated_at", "-updated_at", "start_at", "-start_at"
+    ]
+    | None = Query(
+        None,
+        description="Ordering field",
+    ),
+) -> list[Snapshot]:
+    """Get all snapshots."""
+    return SnapshotModel.objects.filter(status=SnapshotModel.Status.COMPLETED).order_by(
+        ordering or "-created_at"
+    )
+
+
+@router.get(
+    "/{str:snapshot_key}",
+    description="Retrieve snapshot details.",
+    operation_id="get_snapshot",
+    response={
+        HTTPStatus.NOT_FOUND: SnapshotError,
+        HTTPStatus.OK: SnapshotDetail,
+    },
+    summary="Get snapshot",
+)
+@decorate_view(cache_response())
+def get_snapshot(
+    request: HttpRequest,
+    snapshot_key: str = Path(example="2025-02"),
+) -> SnapshotDetail | SnapshotError:
+    """Get snapshot."""
+    if snapshot := SnapshotModel.objects.filter(
+        key__iexact=snapshot_key, status=SnapshotModel.Status.COMPLETED
+    ).first():
+        return snapshot
+
+    return Response({"message": "Snapshot not found"}, status=HTTPStatus.NOT_FOUND)
+
+
+@router.get(
+    "/{str:snapshot_key}/chapters/",
+    description="Retrieve a paginated list of new chapters in a snapshot.",
+    operation_id="list_snapshot_chapters",
+    response=list[Chapter],
+    summary="List new chapters in snapshot",
+)
+@decorate_view(cache_response())
+def list_snapshot_chapters(
+    request: HttpRequest,
+    snapshot_key: str = Path(example="2025-02"),
+    ordering: Literal["created_at", "-created_at", "updated_at", "-updated_at"] | None = Query(
+        None,
+        description="Ordering field",
+    ),
+) -> list[Chapter]:
+    """Get new chapters in snapshot."""
+    if snapshot := SnapshotModel.objects.filter(
+        key__iexact=snapshot_key, status=SnapshotModel.Status.COMPLETED
+    ).first():
+        return snapshot.new_chapters.order_by(ordering or "-created_at")
+
+    return ChapterModel.objects.none()
+
+
+@router.get(
+    "/{str:snapshot_key}/issues/",
+    description="Retrieve a paginated list of new issues in a snapshot.",
+    operation_id="list_snapshot_issues",
+    response=list[Issue],
+    summary="List new issues in snapshot",
+)
+@decorate_view(cache_response())
+def list_snapshot_issues(
+    request: HttpRequest,
+    snapshot_key: str = Path(example="2025-02"),
+    ordering: Literal["created_at", "-created_at", "updated_at", "-updated_at"] | None = Query(
+        None,
+        description="Ordering field",
+    ),
+) -> list[Issue]:
+    """Get new issues in snapshot."""
+    if snapshot := SnapshotModel.objects.filter(
+        key__iexact=snapshot_key, status=SnapshotModel.Status.COMPLETED
+    ).first():
+        return snapshot.new_issues.order_by(ordering or "-created_at")
+    return IssueModel.objects.none()
+
+
+@router.get(
+    "/{str:snapshot_key}/projects/",
+    description="Retrieve a paginated list of new projects in a snapshot.",
+    operation_id="list_snapshot_projects",
+    response=list[Project],
+    summary="List new projects in snapshot",
+)
+@decorate_view(cache_response())
+def list_snapshot_projects(
+    request: HttpRequest,
+    snapshot_key: str = Path(example="2025-02"),
+    ordering: Literal["created_at", "-created_at", "updated_at", "-updated_at"] | None = Query(
+        None,
+        description="Ordering field",
+    ),
+) -> list[Project]:
+    """Get new projects in snapshot."""
+    if snapshot := SnapshotModel.objects.filter(
+        key__iexact=snapshot_key, status=SnapshotModel.Status.COMPLETED
+    ).first():
+        return snapshot.new_projects.order_by(ordering or "-created_at")
+    return ProjectModel.objects.none()
+
+
+@router.get(
+    "/{str:snapshot_key}/releases/",
+    description="Retrieve a paginated list of new releases in a snapshot.",
+    operation_id="list_snapshot_releases",
+    response=list[Release],
+    summary="List new releases in snapshot",
+)
+@decorate_view(cache_response())
+def list_snapshot_releases(
+    request: HttpRequest,
+    snapshot_key: str = Path(example="2025-02"),
+    ordering: Literal["created_at", "-created_at", "published_at", "-published_at"] | None = Query(
+        None,
+        description="Ordering field",
+    ),
+) -> list[Release]:
+    """Get new releases in snapshot."""
+    if snapshot := SnapshotModel.objects.filter(
+        key__iexact=snapshot_key, status=SnapshotModel.Status.COMPLETED
+    ).first():
+        return snapshot.new_releases.order_by(ordering or "-created_at")
+    return ReleaseModel.objects.none()
+
+
+@router.get(
+    "/{str:snapshot_key}/users/",
+    description="Retrieve a paginated list of new users in a snapshot.",
+    operation_id="list_snapshot_users",
+    response=list[Member],
+    summary="List new users in snapshot",
+)
+@decorate_view(cache_response())
+def list_snapshot_users(
+    request: HttpRequest,
+    snapshot_key: str = Path(example="2025-02"),
+    ordering: Literal["created_at", "-created_at", "updated_at", "-updated_at"] | None = Query(
+        None,
+        description="Ordering field",
+    ),
+) -> list[Member]:
+    """Get new users in snapshot."""
+    if snapshot := SnapshotModel.objects.filter(
+        key__iexact=snapshot_key, status=SnapshotModel.Status.COMPLETED
+    ).first():
+        return snapshot.new_users.order_by(ordering or "-created_at")
+    return UserModel.objects.none()

backend/apps/owasp/models/snapshot.py

@@ -45,3 +45,28 @@ def save(self, *args, **kwargs) -> None:
             self.key = now().strftime("%Y-%m")
 
         super().save(*args, **kwargs)
+
+    @property
+    def new_chapters_count(self) -> int:
+        """Return the count of new chapters."""
+        return self.new_chapters.count()
+
+    @property
+    def new_issues_count(self) -> int:
+        """Return the count of new issues."""
+        return self.new_issues.count()
+
+    @property
+    def new_projects_count(self) -> int:
+        """Return the count of new projects."""
+        return self.new_projects.count()
+
+    @property
+    def new_releases_count(self) -> int:
+        """Return the count of new releases."""
+        return self.new_releases.count()
+
+    @property
+    def new_users_count(self) -> int:
+        """Return the count of new users."""
+        return self.new_users.count()