Add endpoints to create and retrieve secret message

futuramaapi/routers/__init__.py

@@ -2,6 +2,7 @@
 
 from .callbacks import router as callbacks_router
 from .characters import router as characters_router
+from .crypto import router as crypto_router
 from .episodes import router as episodes_router
 from .graphql import router as graphql_router
 from .notifications import router as notification_router
@@ -20,6 +21,7 @@
 
 api_router.include_router(callbacks_router)
 api_router.include_router(characters_router)
+api_router.include_router(crypto_router)
 api_router.include_router(episodes_router)
 api_router.include_router(notification_router)
 api_router.include_router(seasons_router)

futuramaapi/routers/crypto/__init__.py

@@ -0,0 +1,5 @@
+from .api import router
+
+__all__ = [
+    "router",
+]

futuramaapi/routers/crypto/api.py

@@ -0,0 +1,62 @@
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from sqlalchemy.ext.asyncio.session import AsyncSession
+
+from futuramaapi.repositories.session import get_async_session
+from futuramaapi.routers.exceptions import ModelNotFoundError, NotFoundResponse
+
+from .schemas import (
+    HiddenSecretMessage,
+    SecretMessage,
+    SecretMessageCreateRequest,
+)
+
+router = APIRouter(
+    prefix="/crypto",
+    tags=["crypto"],
+)
+
+
+@router.post(
+    "/secret_message",
+    status_code=status.HTTP_201_CREATED,
+    response_model=SecretMessage,
+    name="create_secret_message",
+)
+async def create_secret_message(
+    data: SecretMessageCreateRequest,
+    session: AsyncSession = Depends(get_async_session),  # noqa: B008
+) -> SecretMessage:
+    """Create Secret message."""
+    return await SecretMessage.create(
+        session,
+        data,
+        extra_fields={
+            "ip_address": "",
+        },
+    )
+
+
+@router.get(
+    "/secret_message/{url}",
+    status_code=status.HTTP_200_OK,
+    response_model=SecretMessage | HiddenSecretMessage,
+    responses={
+        status.HTTP_404_NOT_FOUND: {
+            "model": NotFoundResponse,
+        },
+    },
+    name="get_secret_message",
+)
+async def get_secret_message(
+    url: str,
+    request: Request,
+    session: AsyncSession = Depends(get_async_session),  # noqa: B008
+) -> SecretMessage | HiddenSecretMessage:
+    """Get Secret message."""
+    try:
+        return await SecretMessage.get_once(session, request, url)
+    except ModelNotFoundError:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Not Found",
+        ) from None

futuramaapi/routers/crypto/schemas.py

@@ -0,0 +1,77 @@
+from operator import add
+from typing import ClassVar, Self
+
+from fastapi import Request
+from pydantic import Field
+from sqlalchemy.ext.asyncio.session import AsyncSession
+
+from futuramaapi.helpers.pydantic import BaseModel
+from futuramaapi.mixins.pydantic import BaseModelDatabaseMixin
+from futuramaapi.repositories.models import SecretMessageModel
+from futuramaapi.routers.exceptions import ModelNotFoundError
+
+
+class SecretMessageCreateRequest(BaseModel):
+    text: str = Field(
+        min_length=1,
+        max_length=8192,
+    )
+
+
+class BaseSecretMessageError(Exception): ...
+
+
+class SecretMessageBase(BaseModel):
+    visit_counter: int
+    id: int
+    ip_address: str
+    url: str
+
+
+class HiddenSecretMessage(SecretMessageBase): ...
+
+
+class SecretMessage(SecretMessageBase, BaseModelDatabaseMixin):
+    model: ClassVar[type[SecretMessageModel]] = SecretMessageModel
+    max_visit_counter: ClassVar[int] = 2
+
+    text: str
+
+    @classmethod
+    async def get_once(
+        cls,
+        session: AsyncSession,
+        request: Request,
+        url: str,
+        /,
+    ) -> Self | HiddenSecretMessage:
+        """
+        3 requests to the DB.
+        I need to forget about creating ORM and use sqlalchemy directly...
+        """
+        try:
+            message: SecretMessage = await SecretMessage.get(
+                session,
+                url,
+                field=SecretMessageModel.url,
+            )
+        except ModelNotFoundError:
+            raise
+
+        await message.update(
+            session,
+            None,
+            visit_counter=add(message.visit_counter, 1),
+            ip_address=request.client.host,
+        )
+
+        if message.visit_counter >= cls.max_visit_counter:
+            return HiddenSecretMessage(**message.model_dump())
+
+        await message.update(
+            session,
+            None,
+            ip_address=request.client.host,
+        )
+
+        return message