Add base controller, catch controller, and history controller

controller/base_controller.py

@@ -0,0 +1,15 @@
+class BaseController:
+    RESTRICTED_NAMESPACES = [
+        "__history",
+        "__last_request",
+        "__last",
+        "__clear",
+        "docs",
+        "redoc",
+        "api",
+    ]
+
+    def check_namespace(self, namespace: str) -> bool:
+        if namespace in self.RESTRICTED_NAMESPACES:
+            return False
+        return True

controller/catch_controller.py

@@ -0,0 +1,36 @@
+from fastapi import Request, Response, status
+from schema.answer import Answer
+from controller.base_controller import BaseController
+from datetime import datetime
+from manager.history import History
+from schema.status import Status
+
+
+class CatchController(BaseController):
+    async def catch(
+        self,
+        request: Request,
+        response: Response,
+        namespace: str = "requests",
+    ) -> Answer:
+        if not self.check_namespace(namespace):
+            response.status_code = status.HTTP_400_BAD_REQUEST
+            return Answer(
+                status=Status.error, message="Invalid namespace name provided."
+            )
+        try:
+            json = await request.json()
+        except Exception:
+            json = {"invalid": "json"}
+
+        last_request = {
+            "data": json,
+            "method": request.method,
+            "url": str(request.url),
+            "headers": dict(request.headers),
+            "time": datetime.now().isoformat(),
+        }
+        history = History(namespace)
+        history.add(last_request)
+
+        return Answer(status=Status.ok, message="Request was catched.")

controller/history_controller.py

@@ -0,0 +1,73 @@
+from fastapi import Response, status
+from schema.answer import Answer
+from schema.request_data import RequestData
+from manager.history import History
+from controller.base_controller import BaseController
+from schema.status import Status
+
+
+class HistoryController(BaseController):
+    async def history(
+        self,
+        response: Response,
+        id: int | None = None,
+        namespace: str = "requests",
+    ) -> Answer | RequestData | list[RequestData]:
+        if not self.check_namespace(namespace=namespace):
+            response.status_code = status.HTTP_400_BAD_REQUEST
+            return Answer(
+                status=Status.error,
+                message="Invalid namespace name provided.",
+            )
+
+        history = History(namespace=namespace)
+
+        if id is not None:
+            item = history.get(id)
+            if item is None:
+                return Answer(
+                    status=Status.error,
+                    message="No requests found.",
+                )
+            return RequestData(**item)
+        return [RequestData(**d) for d in history.all()]
+
+    async def last_requests(
+        self,
+        response: Response,
+        namespace: str = "requests",
+    ) -> Answer | RequestData:
+        if not self.check_namespace(namespace=namespace):
+            response.status_code = status.HTTP_400_BAD_REQUEST
+            return Answer(
+                status=Status.error,
+                message="Invalid namespace name provided.",
+            )
+
+        last = History(namespace=namespace).last()
+
+        if last is None:
+            return Answer(
+                status=Status.error,
+                message="No requests found.",
+            )
+
+        return RequestData(**last)
+
+    async def clear_history(
+        self,
+        response: Response,
+        namespace: str = "requests",
+    ) -> Answer:
+        if not self.check_namespace(namespace=namespace):
+            response.status_code = status.HTTP_400_BAD_REQUEST
+            return Answer(
+                status=Status.error,
+                message="Invalid namespace name provided.",
+            )
+
+        History(namespace=namespace).clear()
+        return Answer(
+            status=Status.success,
+            message="History cleared.",
+        )