feat(backend): support update site_setting (#9)

Author: wd0517

backend/app/alembic/versions/2fc10c21bf88_.py

@@ -0,0 +1,31 @@
+"""empty message
+
+Revision ID: 2fc10c21bf88
+Revises: 8248a83756fd
+Create Date: 2024-07-07 06:14:43.743558
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from tidb_vector.sqlalchemy import VectorType
+
+
+# revision identifiers, used by Alembic.
+revision = "2fc10c21bf88"
+down_revision = "8248a83756fd"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_unique_constraint(None, "site_settings", ["name"])
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, "site_settings", type_="unique")
+    # ### end Alembic commands ###

backend/app/api/admin_routes/site_setting.py

@@ -1,13 +1,66 @@
-from typing import Dict
+from typing import Dict, Union
 from pydantic import BaseModel
-from fastapi import APIRouter
+from http import HTTPStatus
+from fastapi import APIRouter, HTTPException
+from fastapi.exceptions import RequestValidationError
 
-from app.api.deps import CurrentSuperuserDep
-from app.site_settings import SiteSetting, SettingValue
+from app.api.deps import CurrentSuperuserDep, SessionDep
+from app.site_settings import SiteSetting, SettingValue, SettingType
 
 router = APIRouter()
 
 
 @router.get("/admin/site-settings", response_model=Dict[str, SettingValue])
-def site_settings(current_superuser: CurrentSuperuserDep):
-    return SiteSetting.get_all_settings()
+def site_settings(user: CurrentSuperuserDep):
+    return SiteSetting.get_all_settings(force_check_db_cache=True)
+
+
+class SettingUpdate(BaseModel):
+    value: SettingType
+
+
+@router.put(
+    "/admin/site-settings/{setting_name}",
+    status_code=HTTPStatus.NO_CONTENT,
+    responses={
+        HTTPStatus.BAD_REQUEST: {
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "invalid_data_type": {
+                            "summary": "Invalid data type",
+                            "value": {"detail": "title must be of type `str`"},
+                        },
+                    }
+                }
+            },
+        },
+        HTTPStatus.NOT_FOUND: {
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "setting_not_found": {
+                            "summary": "Setting not found",
+                            "value": {"detail": "Setting not found"},
+                        },
+                    }
+                }
+            },
+        },
+    },
+)
+def update_site_setting(
+    session: SessionDep,
+    user: CurrentSuperuserDep,
+    setting_name: str,
+    request: SettingUpdate,
+):
+    if not SiteSetting.setting_exists(setting_name):
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND, detail="Setting not found"
+        )
+
+    try:
+        SiteSetting.update_setting(session, setting_name, request.value)
+    except ValueError as e:
+        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(e))

backend/app/models/site_setting.py

@@ -7,7 +7,7 @@
 
 class SiteSetting(UpdatableBaseModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
-    name: str = Field(max_length=256)
+    name: str = Field(max_length=256, unique=True)
     data_type: str = Field(max_length=256)
     value: str = Field(sa_column=Column(JSON))
 

backend/app/site_settings/__init__.py

@@ -29,21 +29,31 @@ def get_db_last_updated_at(session: Session):
     return result.timestamp() if result else 0
 
 
+type_mapping = {
+    "str": str,
+    "int": int,
+    "float": float,
+    "bool": bool,
+    "dict": dict,
+    "list": list,
+}
+
+
 class SiteSettingProxy:
     __db_cache: dict = {}
     __last_updated_at_ts: float = 0
     __last_checked_at_ts: float = 0
     __mutex = threading.Lock()
 
-    def update_db_cache(self):
+    def update_db_cache(self, force_check=False):
         with Session(engine) as session:
             # Check if we need to update the cache every 10 seconds,
             # so it means settings will not be updated in real-time
             # which is acceptable for this project.
             # If we need real-time updates in the future, we can use
             # a message queue or a pub/sub system to notify the app.
             now = time.time()
-            if now - self.__last_checked_at_ts > 10:
+            if force_check or (now - self.__last_checked_at_ts > 10):
                 self.__last_checked_at_ts = now
                 last_updated_at_ts = get_db_last_updated_at(session)
 
@@ -89,8 +99,10 @@ def get_setting_group(self, group: str) -> dict[str, SettingValue]:
             )
         return result
 
-    def get_all_settings(self) -> dict[str, SettingValue]:
-        self.update_db_cache()
+    def get_all_settings(
+        self, force_check_db_cache: bool = False
+    ) -> dict[str, SettingValue]:
+        self.update_db_cache(force_check_db_cache)
 
         result = {}
         for _, settings in default_settings.setting_groups.items():
@@ -106,8 +118,31 @@ def get_all_settings(self) -> dict[str, SettingValue]:
                 )
         return result
 
+    def setting_exists(self, name: str) -> bool:
+        return hasattr(default_settings, name)
+
+    def update_setting(self, session: Session, name: str, value: SettingType):
+        if not self.setting_exists(name):
+            raise AttributeError(f"Setting {name} does not exist.")
+
+        _default_setting: SettingValue = getattr(default_settings, name)
+        if not isinstance(value, type_mapping[_default_setting.data_type]):
+            raise ValueError(f"{name} must be of type `{_default_setting.data_type}`.")
+
+        db_setting_obj = session.exec(
+            select(DBSiteSetting).filter(DBSiteSetting.name == name)
+        ).first()
+        if db_setting_obj:
+            db_setting_obj.value = value
+        else:
+            db_setting_obj = DBSiteSetting(
+                name=name, value=value, data_type=_default_setting.data_type
+            )
+            session.add(db_setting_obj)
+        session.commit()
+
 
 SiteSetting = SiteSettingProxy()
 
 
-__all__ = ["SiteSetting", "SettingValue"]
+__all__ = ["SiteSetting", "SettingValue", "SettingType"]