feat: added logging API

app/main.py

@@ -77,8 +77,10 @@ app.add_middleware(
 
 # App Routers
 from app.routes.app import router as app_router
+from app.routes.entry import router as entry_router
 
 app.include_router(app_router)
+app.include_router(entry_router)
 
 
 # Pagination

app/models/entry.py

@@ -23,11 +23,15 @@ class LogEntry(Base):
     application = Column(
         UUID(as_uuid=True), ForeignKey("application.id", ondelete="CASCADE"), nullable=False
     )
-
+    # type of the log entry
-    t_type = Column(Enum(TransactionType), nullable=False, default=TransactionType.UNDEFINED)
     l_type = Column(Enum(LogType), nullable=False, default=LogType.INFO)
-
+    # type of the transaction
+    t_type = Column(Enum(TransactionType), nullable=False, default=TransactionType.UNDEFINED)
+    # a custom logmessage
     message = Column(String(512), nullable=True)
+    # author ID i.e. auth0 user sub
     author = Column(String(512), nullable=False, default="system")
-
+    # optional reference to the object (like object ID)
+    object_reference = Column(String(512), nullable=True)
+    # for irreversible operations, store the object state before the operation
     previous_object = Column(JSON, nullable=True)

app/routes/entry.py

@@ -0,0 +1,65 @@
+from creyPY.fastapi.crud import (
+    create_obj_from_data,
+)
+from creyPY.fastapi.db.session import get_db
+from fastapi import APIRouter, Depends, Security, HTTPException
+from sqlalchemy.orm import Session
+
+from app.services.auth import verify
+from app.schema.entry import LogIN, LogOUT
+from app.models.entry import LogEntry
+from fastapi_pagination.ext.sqlalchemy import paginate
+from creyPY.fastapi.pagination import Page
+from uuid import UUID
+
+router = APIRouter(prefix="/log", tags=["logging"])
+
+
+@router.post("/", status_code=201)
+async def create_log(
+    data: LogIN,
+    sub: str = Security(verify),
+    db: Session = Depends(get_db),
+) -> LogOUT:
+    obj = create_obj_from_data(
+        data,
+        LogEntry,
+        db,
+        additonal_data={"created_by_id": sub},
+    )
+    return LogOUT.model_validate(obj)
+
+
+@router.delete("/{log_id}", status_code=204)
+async def delete_log(
+    log_id: UUID,
+    sub: str = Security(verify),
+    db: Session = Depends(get_db),
+) -> None:
+    obj = db.query(LogEntry).filter_by(id=log_id, created_by_id=sub).one_or_none()
+    if obj is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    db.delete(obj)
+    db.commit()
+    return None
+
+
+@router.get("/{log_id}")
+async def get_log(
+    log_id: UUID,
+    sub: str = Security(verify),
+    db: Session = Depends(get_db),
+) -> LogOUT:
+    obj = db.query(LogEntry).filter_by(id=log_id, created_by_id=sub).one_or_none()
+    if obj is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return LogOUT.model_validate(obj)
+
+
+@router.get("/")
+async def get_logs(
+    sub: str = Security(verify),
+    db: Session = Depends(get_db),
+) -> Page[LogOUT]:
+    the_select = db.query(LogEntry).filter_by(created_by_id=sub)
+    return paginate(the_select)

app/schema/entry.py

@@ -0,0 +1,19 @@
+from app.schema.common import BaseSchemaModelIN, BaseSchemaModelOUT
+from app.models.entry import TransactionType, LogType
+from uuid import UUID
+from pydantic.json_schema import SkipJsonSchema
+
+
+class LogIN(BaseSchemaModelIN):
+    application: UUID
+    l_type: LogType = LogType.INFO
+    t_type: TransactionType = TransactionType.UNDEFINED
+
+    message: str | SkipJsonSchema[None] = None
+    author: str = "system"
+    object_reference: str | SkipJsonSchema[None] = None
+    previous_object: dict | SkipJsonSchema[None] = None
+
+
+class LogOUT(BaseSchemaModelOUT, LogIN):
+    pass

app/test_main.py

@@ -6,12 +6,21 @@ from sqlalchemy.orm import sessionmaker
 from sqlalchemy_utils import create_database, database_exists, drop_database
 
 from app.services.auth import verify
-
+import contextlib
 from .main import app
 
 CURRENT_USER = "api-key|testing"
 
 
+@contextlib.contextmanager
+def app_context(self):
+    app_id = self.create_app()
+    try:
+        yield app_id
+    finally:
+        self.destroy_app(app_id)
+
+
 class TestAPI:
     def setup_class(self):
         self.engine = create_engine(SQLALCHEMY_DATABASE_URL + "test", pool_pre_ping=True)
@@ -34,7 +43,6 @@ class TestAPI:
 
         app.dependency_overrides[get_db] = get_db_test
         app.dependency_overrides[verify] = get_test_sub
-
         self.c = GenericClient(app)
 
     def teardown_class(self):
@@ -49,3 +57,29 @@ class TestAPI:
 
     def test_application_api(self):
         self.c.obj_lifecycle({"name": "Testing"}, "/app/")
+
+    def create_app(self):
+        re = self.c.post("/app/", {"name": "Testing"})
+        return re["id"]
+
+    def destroy_app(self, app_id):
+        self.c.delete(f"/app/{app_id}")
+
+    def test_log_api(self):
+        with app_context(self) as app_id:
+            self.c.obj_lifecycle({"application": app_id}, "/log/")
+
+    def test_logging_standards(self):
+        with app_context(self) as app_id:
+            re = self.c.post("/log/", {"application": app_id})
+            log_id = re["id"]
+            assert re["application"] == app_id
+            assert re["l_type"] == "info"
+            assert re["t_type"] == "undefined"
+            assert re["message"] == None
+            assert re["author"] == "system"
+            assert re["object_reference"] == None
+            assert re["previous_object"] == None
+            assert re["created_by_id"] == CURRENT_USER
+
+            self.c.delete(f"/log/{log_id}")