feat: Add announcements endpoints (#91)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Alexandre Teles (afterSt0rm) <alexandre.teles@ufba.br>

api/__init__.py

@@ -7,5 +7,9 @@ from api.socials import socials
 from api.info import info
 from api.compat import github as compat
 from api.donations import donations
+from api.announcements import announcements
+from api.login import login
 
-api = Blueprint.group(ping, github, info, socials, donations, compat, url_prefix="/")
+api = Blueprint.group(
+    login, ping, github, info, socials, donations, announcements, compat, url_prefix="/"
+)

api/announcements.py

@@ -0,0 +1,242 @@
+"""
+This module provides a blueprint for the announcements endpoint.
+
+Routes:
+    - GET /announcements: Get a list of announcements from all channels.
+    - GET /announcements/<channel:str>: Get a list of announcement from a channel.
+    - GET /announcements/latest: Get the latest announcement.
+    - GET /announcements/<channel:str>/latest: Get the latest announcement from a channel.
+    - POST /announcements/<channel:str>: Create an announcement.
+    - DELETE /announcements/<announcement_id:int>: Delete an announcement.
+"""
+
+import datetime
+from sanic import Blueprint, Request
+from sanic.response import JSONResponse, json
+from sanic_ext import openapi
+from data.database import Session
+from data.models import AnnouncementDbModel, AttachmentDbModel
+
+import sanic_beskar
+
+from api.models.announcements import AnnouncementResponseModel
+from config import api_version
+from limiter import limiter
+
+announcements: Blueprint = Blueprint("announcements", version=api_version)
+
+
+@announcements.get("/announcements")
+@openapi.definition(
+    summary="Get a list of announcements",
+    response=[[AnnouncementResponseModel]],
+)
+async def get_announcements(request: Request) -> JSONResponse:
+    """
+    Retrieve a list of announcements.
+
+    **Returns:**
+        - JSONResponse: A Sanic JSONResponse object containing a list of announcements from all channels.
+    """
+
+    session = Session()
+
+    announcements = [
+        AnnouncementResponseModel.to_response(announcement)
+        for announcement in session.query(AnnouncementDbModel).all()
+    ]
+
+    session.close()
+
+    return json(announcements, status=200)
+
+
+@announcements.get("/announcements/<channel:str>")
+@openapi.definition(
+    summary="Get a list of announcements from a channel",
+    response=[[AnnouncementResponseModel]],
+)
+async def get_announcements_for_channel(request: Request, channel: str) -> JSONResponse:
+    """
+    Retrieve a list of announcements from a channel.
+
+    **Args:**
+        - channel (str): The channel to retrieve announcements from.
+
+    **Returns:**
+        - JSONResponse: A Sanic JSONResponse object containing a list of announcements from a channel.
+    """
+
+    session = Session()
+
+    announcements = [
+        AnnouncementResponseModel.to_response(announcement)
+        for announcement in session.query(AnnouncementDbModel)
+        .filter_by(channel=channel)
+        .all()
+    ]
+
+    session.close()
+
+    return json(announcements, status=200)
+
+
+@announcements.get("/announcements/latest")
+@openapi.definition(
+    summary="Get the latest announcement",
+    response=AnnouncementResponseModel,
+)
+async def get_latest_announcement(request: Request) -> JSONResponse:
+    """
+    Retrieve the latest announcement.
+
+    **Returns:**
+        - JSONResponse: A Sanic JSONResponse object containing the latest announcement.
+    """
+
+    session = Session()
+
+    announcement = (
+        session.query(AnnouncementDbModel)
+        .order_by(AnnouncementDbModel.id.desc())
+        .first()
+    )
+
+    if not announcement:
+        return json({"error": "No announcement found"}, status=404)
+
+    announcement_response = AnnouncementResponseModel.to_response(announcement)
+
+    session.close()
+
+    return json(announcement_response, status=200)
+
+
+# for specific channel
+
+
+@announcements.get("/announcements/<channel:str>/latest")
+@openapi.definition(
+    summary="Get the latest announcement from a channel",
+    response=AnnouncementResponseModel,
+)
+async def get_latest_announcement_for_channel(
+    request: Request, channel: str
+) -> JSONResponse:
+    """
+    Retrieve the latest announcement from a channel.
+
+    **Args:**
+        - channel (str): The channel to retrieve the latest announcement from.
+
+    **Returns:**
+        - JSONResponse: A Sanic JSONResponse object containing the latest announcement from a channel.
+    """
+
+    session = Session()
+
+    announcement = (
+        session.query(AnnouncementDbModel)
+        .filter_by(channel=channel)
+        .order_by(AnnouncementDbModel.id.desc())
+        .first()
+    )
+
+    if not announcement:
+        return json({"error": "No announcement found"}, status=404)
+
+    announcement_response = AnnouncementResponseModel.to_response(announcement)
+
+    session.close()
+
+    return json(announcement_response, status=200)
+
+
+@announcements.post("/announcements/<channel:str>")
+@limiter.limit("16 per hour")
+@sanic_beskar.auth_required
+@openapi.definition(
+    summary="Create an announcement",
+    body=AnnouncementResponseModel,
+    response=AnnouncementResponseModel,
+)
+async def post_announcement(request: Request, channel: str) -> JSONResponse:
+    """
+    Create an announcement.
+
+    **Args:**
+        - author (str | None): The author of the announcement.
+        - title (str): The title of the announcement.
+        - content (ContentFields | None): The content of the announcement.
+        - channel (str): The channel to create the announcement in.
+        - nevel (int | None): The severity of the announcement.
+    """
+    session = Session()
+
+    if not request.json:
+        return json({"error": "Missing request body"}, status=400)
+
+    content = request.json.get("content", None)
+
+    author = request.json.get("author", None)
+    title = request.json.get("title")
+    message = content["message"] if content and "message" in content else None
+    attachments = (
+        list(
+            map(
+                lambda url: AttachmentDbModel(attachment_url=url),
+                content["attachments"],
+            )
+        )
+        if content and "attachments" in content
+        else []
+    )
+    level = request.json.get("level", None)
+    created_at = datetime.datetime.now()
+
+    announcement = AnnouncementDbModel(
+        author=author,
+        title=title,
+        message=message,
+        attachments=attachments,
+        channel=channel,
+        created_at=created_at,
+        level=level,
+    )
+
+    session.add(announcement)
+    session.commit()
+    session.close()
+
+    return json({}, status=200)
+
+
+@announcements.delete("/announcements/<announcement_id:int>")
+@sanic_beskar.auth_required
+@openapi.definition(
+    summary="Delete an announcement",
+)
+async def delete_announcement(request: Request, announcement_id: int) -> JSONResponse:
+    """
+    Delete an announcement.
+
+    **Args:**
+        - announcement_id (int): The ID of the announcement to delete.
+
+    **Exceptions:**
+        - 404: Announcement not found.
+    """
+    session = Session()
+
+    announcement = (
+        session.query(AnnouncementDbModel).filter_by(id=announcement_id).first()
+    )
+
+    if not announcement:
+        return json({"error": "Announcement not found"}, status=404)
+
+    session.delete(announcement)
+    session.commit()
+    session.close()
+
+    return json({}, status=200)

api/login.py

@@ -0,0 +1,54 @@
+"""
+This module provides a blueprint for the login endpoint.
+
+Routes:
+    - POST /login: Login to the API
+"""
+
+from sanic import Blueprint, Request
+from sanic.response import JSONResponse, json
+from sanic_ext import openapi
+from sanic_beskar.exceptions import AuthenticationError
+
+from auth import beskar
+from limiter import limiter
+
+from config import api_version
+
+
+login: Blueprint = Blueprint("login", version=api_version)
+
+
+@login.post("/login")
+@openapi.definition(
+    summary="Login to the API",
+)
+@limiter.limit("3 per hour")
+async def login_user(request: Request) -> JSONResponse:
+    """
+    Login to the API.
+
+    **Args:**
+        - username (str): The username of the user to login.
+        - password (str): The password of the user to login.
+
+    **Returns:**
+        - JSONResponse: A Sanic JSONResponse object containing the access token.
+    """
+
+    req = request.json
+    username = req.get("username", None)
+    password = req.get("password", None)
+    if not username or not password:
+        return json({"error": "Missing username or password"}, status=400)
+
+    try:
+        user = await beskar.authenticate(username, password)
+    except AuthenticationError:
+        return json({"error": "Invalid username or password"}, status=403)
+
+    if not user:
+        return json({"error": "Invalid username or password"}, status=403)
+
+    ret = {"access_token": await beskar.encode_token(user)}
+    return json(ret, status=200)

api/models/announcements.py

@@ -0,0 +1,37 @@
+from data.models import AnnouncementDbModel
+
+
+class ContentFields(dict):
+    message: str | None
+    attachment_urls: list[str] | None
+
+
+class AnnouncementResponseModel(dict):
+    id: int
+    author: str | None
+    title: str
+    content: ContentFields | None
+    channel: str
+    created_at: str
+    level: int | None
+
+    @staticmethod
+    def to_response(announcement: AnnouncementDbModel):
+        response = AnnouncementResponseModel(
+            id=announcement.id,
+            author=announcement.author,
+            title=announcement.title,
+            content=ContentFields(
+                message=announcement.message,
+                attachment_urls=[
+                    attachment.attachment_url for attachment in announcement.attachments
+                ],
+            )
+            if announcement.message or announcement.attachments
+            else None,
+            channel=announcement.channel,
+            created_at=str(announcement.created_at),
+            level=announcement.level,
+        )
+
+        return response