Polishing schemas endpoint only including the get_all_schemas method

Clean up SchemaDefinition

Added condition for uniqueness of title in schemas_collection

Author: JosePizarro3

jsoned/api/api_v1/api.py

@@ -1,7 +1,8 @@
-# TODO add login with oauth2
 from fastapi import APIRouter
 
 from jsoned.api.api_v1.endpoints import schemas
 
 api_router = APIRouter()
 api_router.include_router(schemas.router, prefix="/schemas", tags=["schemas"])
+# TODO add login with oauth2
+# api_router.include_router(login.router, prefix="/login", tags=["login"])

jsoned/api/api_v1/endpoints/schemas.py

@@ -1,9 +1,4 @@
-from datetime import datetime
-from typing import Any
-from uuid import uuid4
-
-from fastapi import APIRouter, HTTPException
-from fastapi.encoders import jsonable_encoder
+from fastapi import APIRouter
 
 from jsoned.database import schemas_collection
 from jsoned.models.schema_definition import SchemaDefinition
@@ -12,89 +7,79 @@
 
 
 @router.get("/all", response_model=list[SchemaDefinition])
-async def get_all_schemas() -> list[dict[str, Any]]:
-    """
-    Retrieve all schemas. Ensures each document has `id` as a string and a valid `updated_at`.
-    """
-    docs = list(schemas_collection.find())
-    normalized: list[dict[str, Any]] = []
-
-    for d in docs:
-        # Ensure `id` exists and is a non-empty string
-        if not isinstance(d.get("id"), str) or not d["id"].strip():
-            d["id"] = str(uuid4())
-
-        # Ensure `updated_at` is present (optional safeguard)
-        if d.get("updated_at") is None:
-            d["updated_at"] = datetime.utcnow()
-
-        normalized.append(d)
-
-    # No BSON present; plain JSON encoding is fine
-    return jsonable_encoder(normalized)
-
-
-@router.post("/add", response_model=SchemaDefinition)
-async def add_schema(schema: SchemaDefinition) -> dict[str, Any]:
+async def get_all_schemas():
     """
-    Add a new schema. If `id` is missing/empty, generate one; always refresh `updated_at`.
+    Gets all the schemas present in the database and returns them in a list of `SchemaDefinition` models.
     """
-    doc = schema.dict()
-
-    # Guarantee a usable id
-    if not isinstance(doc.get("id"), str) or not doc["id"].strip():
-        doc["id"] = str(uuid4())
-
-    # Server-side timestamp
-    doc["updated_at"] = datetime.utcnow()
-
-    # Insert as-is (no ObjectId conversions)
-    schemas_collection.insert_one(doc)
-
-    # Return exactly what we stored
-    return jsonable_encoder(doc)
-
-
-@router.put("/update/{id}", response_model=dict[str, str])
-async def update_schema(id: str, update: SchemaDefinition) -> dict[str, str]:
-    """
-    Update schema by `id`. Ignores `id` field in the payload (primary key is immutable).
-    Only non-None fields are updated; `updated_at` is refreshed automatically.
-    """
-    if not isinstance(id, str) or not id.strip():
-        raise HTTPException(
-            status_code=400, detail="Invalid schema id (must be a non-empty string)"
-        )
-
-    # Ignore None values and prevent changing the primary key
-    update_fields = {
-        k: v for k, v in update.dict().items() if v is not None and k != "id"
-    }
-
-    if update_fields:
-        update_fields["updated_at"] = datetime.utcnow()
-
-    result = schemas_collection.update_one(
-        {"id": id}, {"$set": update_fields} if update_fields else {}
-    )
-    if result.matched_count == 0:
-        raise HTTPException(status_code=404, detail="Schema not found")
-
-    return {"message": "Schema updated"}
-
-
-@router.delete("/delete/{id}", response_model=dict[str, str])
-async def delete_schema(id: str) -> dict[str, str]:
-    """
-    Delete schema by `id`.
-    """
-    if not isinstance(id, str) or not id.strip():
-        raise HTTPException(
-            status_code=400, detail="Invalid schema id (must be a non-empty string)"
-        )
-
-    result = schemas_collection.delete_one({"id": id})
-    if result.deleted_count == 0:
-        raise HTTPException(status_code=404, detail="Schema not found")
-
-    return {"message": "Schema deleted"}
+    response = []
+    documents = schemas_collection.find({})
+    async for doc in documents:
+        response.append(SchemaDefinition(**doc))
+
+    return response
+
+
+# @router.post("/add", response_model=SchemaDefinition)
+# async def add_schema(schema: SchemaDefinition) -> dict[str, Any]:
+#     """
+#     Add a new schema. If `id` is missing/empty, generate one; always refresh `updated_at`.
+#     """
+#     doc = schema.dict()
+
+#     # Guarantee a usable id
+#     if not isinstance(doc.get("id"), str) or not doc["id"].strip():
+#         doc["id"] = str(uuid4())
+
+#     # Server-side timestamp
+#     doc["updated_at"] = datetime.utcnow()
+
+#     # Insert as-is (no ObjectId conversions)
+#     schemas_collection.insert_one(doc)
+
+#     # Return exactly what we stored
+#     return jsonable_encoder(doc)
+
+
+# @router.put("/update/{id}", response_model=dict[str, str])
+# async def update_schema(id: str, update: SchemaDefinition) -> dict[str, str]:
+#     """
+#     Update schema by `id`. Ignores `id` field in the payload (primary key is immutable).
+#     Only non-None fields are updated; `updated_at` is refreshed automatically.
+#     """
+#     if not isinstance(id, str) or not id.strip():
+#         raise HTTPException(
+#             status_code=400, detail="Invalid schema id (must be a non-empty string)"
+#         )
+
+#     # Ignore None values and prevent changing the primary key
+#     update_fields = {
+#         k: v for k, v in update.dict().items() if v is not None and k != "id"
+#     }
+
+#     if update_fields:
+#         update_fields["updated_at"] = datetime.utcnow()
+
+#     result = schemas_collection.update_one(
+#         {"id": id}, {"$set": update_fields} if update_fields else {}
+#     )
+#     if result.matched_count == 0:
+#         raise HTTPException(status_code=404, detail="Schema not found")
+
+#     return {"message": "Schema updated"}
+
+
+# @router.delete("/delete/{id}", response_model=dict[str, str])
+# async def delete_schema(id: str) -> dict[str, str]:
+#     """
+#     Delete schema by `id`.
+#     """
+#     if not isinstance(id, str) or not id.strip():
+#         raise HTTPException(
+#             status_code=400, detail="Invalid schema id (must be a non-empty string)"
+#         )
+
+#     result = schemas_collection.delete_one({"id": id})
+#     if result.deleted_count == 0:
+#         raise HTTPException(status_code=404, detail="Schema not found")
+
+#     return {"message": "Schema deleted"}

jsoned/database.py

@@ -11,3 +11,6 @@
 client = AsyncIOMotorClient("mongodb://localhost:27017")
 database = client.jsoned_db
 schemas_collection = database.schemas
+
+# Ensure that the `title` field is unique
+schemas_collection.create_index([("title", 1)], unique=True)

jsoned/models/schema_definition.py

@@ -1,19 +1,27 @@
 from datetime import datetime
+from uuid import UUID
 
 from pydantic import BaseModel, Field
 
 
 class SchemaDefinition(BaseModel):
-    id: str = Field(..., description="Unique identifier for the schema")
-    name: str | None = Field(
+    id: UUID = Field(
+        ...,
+        description="Automatically generated unique identifier for the schema entry based on `content`.",
+    )
+
+    title: str | None = Field(
         None,
-        description="Human-readable name of the schema",
+        description="A human-readable title given to the schema entry.",
         min_length=3,
     )
-    version: str | None = Field(None, description="Version of the schema")
-    content: dict | None = Field(
-        None, description="The actual schema content as a dictionary"
-    )
+
     updated_at: datetime | None = Field(
-        None, description="Timestamp of the last update"
+        None,
+        description="Timestamp of the last update of the schema entry.",
+    )
+
+    content: dict | None = Field(
+        None,
+        description="The actual schema content as a dictionary",
     )

pyproject.toml

@@ -48,7 +48,7 @@ dev = [
 
 [tool.ruff]
 # Exclude a variety of commonly ignored directories.
-include = ["backend/*.py", "tests/*.py"]
+include = ["jsoned/*.py", "tests/*.py"]
 exclude = [
     ".bzr",
     ".direnv",
@@ -128,5 +128,5 @@ where = ["."]
 namespaces = false
 
 [tool.setuptools_scm]
-write_to = "backend/_version.py"
+write_to = "jsoned/_version.py"