Alpha/task24 - Added DB Logger

app/api/endpoints/log_item.py

@@ -0,0 +1,106 @@
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from pymongo.database import Database
+
+from app.api.custom import RouteErrorHandler
+from app.api.deps import get_db, get_user
+from app.crud.log_item import LogItemCRUD
+from app.schemas.common.api_response import ApiGeneralResponse
+from app.schemas.common.object_id import PyObjectId
+from app.schemas.log_item import LogItem, LogItemRead
+from app.schemas.user import UserRead
+
+router = APIRouter(route_class=RouteErrorHandler, tags=["Log Item"])
+
+
+@router.post("/create")
+def create_log_item(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+    log_item: LogItem,
+) -> PyObjectId:
+    crud = LogItemCRUD(db=db)
+    log_item.user_id = current_user.id
+    return crud.create(obj=log_item)
+
+
+@router.get("/get/{obj_id}")
+def read_log_item(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+    obj_id: str,
+) -> LogItemRead:
+    obj_read = LogItemCRUD(db=db).get(obj_id=PyObjectId(obj_id))
+    if obj_read is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="object not found")
+    if obj_read.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden")
+    return obj_read
+
+
+@router.get("/list/{task_id}")
+def list_log_items(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+    task_id: str,
+) -> list[LogItemRead]:
+    return LogItemCRUD(db=db).search(
+        {
+            LogItem.Field.user_id: current_user.id,
+            LogItem.Field.task_id: PyObjectId(task_id),
+        }
+    )
+
+
+@router.get("/count")
+def count_log_items(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+) -> int:
+    return LogItemCRUD(db=db).count({LogItem.Field.user_id: current_user.id})
+
+
+@router.put("/update/{obj_id}")
+def update_log_item(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+    obj_id: str,
+    obj_update: LogItem,
+) -> ApiGeneralResponse:
+    task_crud = LogItemCRUD(db=db)
+    obj_read = task_crud.get(obj_id=PyObjectId(obj_id))
+    if obj_read is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Object not found")
+    if obj_read.user_id != current_user.id:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden")
+    modified_count = task_crud.update(
+        obj_id=PyObjectId(obj_id),
+        obj=obj_update,
+    )
+    return ApiGeneralResponse(
+        message="Object updated",
+        detail={
+            "modified_count": modified_count,
+        },
+    )
+
+
+@router.delete("/delete/{obj_id}")
+def delete_log_item(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+    obj_id: str,
+) -> ApiGeneralResponse:
+    deleted_count = LogItemCRUD(db=db).delete(
+        {
+            "_id": PyObjectId(obj_id),
+            LogItem.Field.user_id: current_user.id,
+        }
+    )
+    return ApiGeneralResponse(
+        message="object deleted",
+        detail={
+            "deleted_count": deleted_count,
+        },
+    )

app/api/endpoints/logs.py

@@ -0,0 +1,32 @@
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends
+from pymongo.database import Database
+
+from app.api.custom import RouteErrorHandler
+from app.api.deps import get_db, get_user
+from app.core.config.api import config as config_api
+from app.crud.log_item import LogItemCRUD
+from app.schemas.common.object_id import PyObjectId
+from app.schemas.log_item import LogItem
+from app.schemas.user import UserRead
+
+router = APIRouter(route_class=RouteErrorHandler, tags=["Logs"])
+
+
+@router.get("/get/{task_id}")
+async def get_logs(
+    db: Annotated[Database[dict[str, Any]], Depends(get_db)],
+    current_user: Annotated[UserRead, Depends(get_user)],
+    task_id: str,
+) -> str:
+    ret: str = ""
+    log_items = LogItemCRUD(db=db).search(
+        {
+            LogItem.Field.user_id: current_user.id,
+            LogItem.Field.task_id: PyObjectId(task_id),
+        }
+    )
+    for log_item in log_items[-config_api.MAX_LOG_LINES :]:
+        ret += f"[{log_item.created_at.isoformat()}] {log_item.content}\n"
+    return ret

app/api/endpoints/scheduler.py

@@ -27,7 +27,7 @@ def create_task(
     obj_list = task_config_crud.search(
         {
             "_id": task_config_id,
-            TaskConfig.FieldName.user_id: current_user.id,
+            TaskConfig.Field.user_id: current_user.id,
         }
     )
     if len(obj_list) == 0:
@@ -67,7 +67,7 @@ def list_tasks(
     task_config_list = TaskConfigCRUD(db=db).search(
         {
             "_id": {"$in": job_id_list},
-            TaskConfig.FieldName.user_id: current_user.id,
+            TaskConfig.Field.user_id: current_user.id,
         }
     )
     for task_config in task_config_list:

app/api/endpoints/task_config.py

@@ -46,7 +46,7 @@ def list_task_configs(
     db: Annotated[Database[dict[str, Any]], Depends(get_db)],
     current_user: Annotated[UserRead, Depends(get_user)],
 ) -> list[TaskConfigRead]:
-    obj_list = TaskConfigCRUD(db=db).search({TaskConfig.FieldName.user_id: current_user.id})
+    obj_list = TaskConfigCRUD(db=db).search({TaskConfig.Field.user_id: current_user.id})
     for obj in obj_list:
         job = scheduler.get_job(job_id=str(obj.id))
         obj.update_with_apjob(job)
@@ -58,7 +58,7 @@ def count_task_configs(
     db: Annotated[Database[dict[str, Any]], Depends(get_db)],
     current_user: Annotated[UserRead, Depends(get_user)],
 ) -> int:
-    return TaskConfigCRUD(db=db).count({TaskConfig.FieldName.user_id: current_user.id})
+    return TaskConfigCRUD(db=db).count({TaskConfig.Field.user_id: current_user.id})
 
 
 @router.put("/update/{obj_id}")
@@ -95,7 +95,7 @@ def delete_task_config(
     deleted_count = TaskConfigCRUD(db=db).delete(
         {
             "_id": PyObjectId(obj_id),
-            TaskConfig.FieldName.user_id: current_user.id,
+            TaskConfig.Field.user_id: current_user.id,
         }
     )
     return ApiGeneralResponse(

app/api/router.py

@@ -5,6 +5,8 @@
 from app.api.custom import RouteErrorHandler
 from app.api.deps import get_user
 from app.api.endpoints.hello import router as router_hello
+from app.api.endpoints.log_item import router as router_log_item
+from app.api.endpoints.logs import router as router_logs
 from app.api.endpoints.scheduler import router as router_scheduler
 from app.api.endpoints.task import router as router_task
 from app.api.endpoints.task_config import router as router_task_config
@@ -19,10 +21,14 @@ class Prefix(StrEnum):
     task_config = "/task-config"
     task = "/task"
     scheduler = "/scheduler"
+    log_item = "/log-item"
+    logs = "/logs"
 
 
 router.include_router(router=router_hello, prefix=Prefix.hello.value)
 router.include_router(router=router_user, prefix=Prefix.user.value)
 router.include_router(router=router_task_config, prefix=Prefix.task_config.value, dependencies=[Depends(get_user)])
 router.include_router(router=router_task, prefix=Prefix.task.value, dependencies=[Depends(get_user)])
 router.include_router(router=router_scheduler, prefix=Prefix.scheduler.value, dependencies=[Depends(get_user)])
+router.include_router(router=router_log_item, prefix=Prefix.log_item.value, dependencies=[Depends(get_user)])
+router.include_router(router=router_logs, prefix=Prefix.logs.value, dependencies=[Depends(get_user)])

app/core/config/api.py

@@ -6,6 +6,7 @@
 
 class Config(BaseSettings):
     DEBUG: bool = config_env.APP_ENV is AppEnv.DEV
+    MAX_LOG_LINES: int = 1000
 
 
 config = Config()

app/crud/log_item.py

@@ -0,0 +1,40 @@
+from typing import Any
+
+from pymongo.database import Database
+
+from app.schemas.common.object_id import PyObjectId
+from app.schemas.log_item import COLLECTION_NAME, LogItem, LogItemRead
+
+
+class LogItemCRUD:
+    def __init__(self, db: Database[dict[str, Any]]) -> None:
+        self.db = db
+        self.coll = db.get_collection(COLLECTION_NAME)
+
+    def create(self, obj: LogItem) -> PyObjectId:
+        res = self.coll.insert_one(obj.model_dump())
+        return PyObjectId(res.inserted_id)
+
+    def get(self, obj_id: PyObjectId) -> LogItemRead | None:
+        ret = None
+        doc = self.coll.find_one({"_id": obj_id})
+        if doc is not None:
+            ret = LogItemRead.model_validate(doc)
+        return ret
+
+    def search(self, filter_by: dict[str, Any] | None = None) -> list[LogItemRead]:
+        cursor = self.coll.find(filter_by) if filter_by is not None else self.coll.find()
+        return [LogItemRead.model_validate(doc) for doc in cursor]
+
+    def count(self, filter_by: dict[str, Any] | None = None) -> int:
+        return self.coll.count_documents(filter_by or {})
+
+    def update(self, obj_id: PyObjectId, obj: LogItem) -> int:
+        res = self.coll.update_one(
+            {"_id": obj_id},
+            {"$set": obj.model_dump()},
+        )
+        return res.modified_count
+
+    def delete(self, filter_by: dict[str, Any] | None = None) -> int:
+        return self.coll.delete_many(filter_by or {}).deleted_count

app/crud/user.py

@@ -40,7 +40,7 @@ def update(self, user_id: PyObjectId, user_update: UserUpdate) -> int:
                 hashed_password = user_org.hashed_password
 
             doc = user_update.model_dump()
-            doc[User.FieldName.hashed_password.value] = hashed_password
+            doc[User.Field.hashed_password.value] = hashed_password
             user_save = User.model_validate(doc)
 
             res = self.coll.update_one({"_id": user_id}, {"$set": user_save.model_dump()})
@@ -53,14 +53,14 @@ def delete(self, filter_by: dict[str, Any] | None = None) -> int:
 
     def get_user_by_username(self, username: str) -> UserRead | None:
         ret = None
-        doc = self.coll.find_one({User.FieldName.username.value: username})
+        doc = self.coll.find_one({User.Field.username.value: username})
         if doc is not None:
             ret = UserRead.model_validate(doc)
         return ret
 
     def get_existing_user(self, user_create: UserCreate) -> User | None:
         ret = None
-        doc = self.coll.find_one({User.FieldName.username.value: user_create.username})
+        doc = self.coll.find_one({User.Field.username.value: user_create.username})
         if doc is not None:
             ret = User.model_validate(doc)
         return ret

app/schemas/configs/enigx.py

@@ -8,12 +8,12 @@ class EnigxConfig(BaseModel):
     project_id: str
     bearer_token: str
 
-    class FieldName(StrEnum):
+    class Field(StrEnum):
         tenant_id = "tenant_id"
         project_id = "project_id"
         bearer_token = "bearer_token"  # noqa: S105
 
-    @field_validator(FieldName.bearer_token.value, mode="before")
+    @field_validator(Field.bearer_token.value, mode="before")
     @classmethod
     def validate_bearer_token(cls, value: str) -> str:
         if not value.startswith("Bearer "):

app/schemas/configs/fetch.py

@@ -47,7 +47,7 @@ class FetchConfig(BaseModel):
     data_type: FetchDataType
     success_code: int = 200
 
-    class FieldName(StrEnum):
+    class Field(StrEnum):
         method = "method"
         url = "url"
         auth_token = "auth_token"  # noqa: S105

app/schemas/configs/task.py

@@ -26,7 +26,7 @@ class TaskConfig(BaseModel):
     interval_secs: int | None = None
     task_args: dict[str, str] = {}
 
-    class FieldName(StrEnum):
+    class Field(StrEnum):
         user_id = "user_id"
         task_name = "task_name"
         task_type = "task_type"
@@ -36,7 +36,7 @@ class FieldName(StrEnum):
         interval_secs = "interval_secs"
         task_args = "task_args"
 
-    @field_validator(FieldName.interval_secs.value, mode="before")
+    @field_validator(Field.interval_secs.value, mode="before")
     @classmethod
     def validate_interval_secs(cls, value: int | None) -> int | None:
         if value is not None and value <= 0:

app/schemas/log_item.py

@@ -0,0 +1,32 @@
+from datetime import UTC, datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, Field, field_validator
+
+from app.schemas.common.object_id import PyObjectId
+
+COLLECTION_NAME = "log_items"
+
+
+class LogItem(BaseModel):
+    user_id: PyObjectId | None
+    task_id: PyObjectId | None
+    content: str
+    created_at: datetime
+
+    class Field(StrEnum):
+        user_id = "user_id"
+        task_id = "task_id"
+        content = "content"
+        created_at = "created_at"
+
+    @field_validator(Field.created_at.value, mode="after")
+    @classmethod
+    def validate_created_at_timezone(cls, value: datetime) -> datetime:
+        if value.tzinfo is None:
+            return value.replace(tzinfo=UTC)
+        return value
+
+
+class LogItemRead(LogItem):
+    id: PyObjectId = Field(alias="_id")

app/schemas/user.py

@@ -18,7 +18,7 @@ class User(BaseModel):
     hashed_password: str
     role: UserRole
 
-    class FieldName(StrEnum):
+    class Field(StrEnum):
         username = "username"
         password = "password"  # noqa: S105
         hashed_password = "hashed_password"  # noqa: S105

app/services/db_logger.py

@@ -0,0 +1,25 @@
+from datetime import UTC, datetime
+from typing import Any
+
+from pymongo.database import Database
+
+from app.crud.log_item import LogItemCRUD
+from app.schemas.common.object_id import PyObjectId
+from app.schemas.log_item import LogItem
+
+
+class DBLogger:
+    def __init__(self, db: Database[dict[str, Any]], user_id: PyObjectId, task_id: PyObjectId) -> None:
+        self.user_id = user_id
+        self.task_id = task_id
+        self.log_crud = LogItemCRUD(db=db)
+
+    def log(self, content: str) -> PyObjectId:
+        return self.log_crud.create(
+            obj=LogItem(
+                user_id=self.user_id,
+                task_id=self.task_id,
+                content=content,
+                created_at=datetime.now(tz=UTC),
+            )
+        )