Multilingual db update

core_backend/Makefile

@@ -35,9 +35,9 @@ run-tests:
 	python -m pytest -rPQ -m "not rails"
 
 # Test DBs
-setup-test-containers: setup-test-db setup-alignscore-container
+setup-test-containers: setup-test-db #setup-alignscore-container
 
-teardown-test-containers: stop-test-db stop-alignscore-container
+teardown-test-containers: stop-test-db #stop-alignscore-container
 
 setup-test-db:
 	-@docker stop testdb

core_backend/app/__init__.py

@@ -8,7 +8,7 @@
     multiprocess,
 )
 
-from . import admin, auth, contents, question_answer, whatsapp_qa
+from . import admin, auth, contents, languages, question_answer, whatsapp_qa
 from .config import (
     DOMAIN,
 )
@@ -33,6 +33,7 @@ def create_app() -> FastAPI:
     app.include_router(contents.router)
     app.include_router(auth.router)
     app.include_router(whatsapp_qa.router)
+    app.include_router(languages.router)
 
     origins = [
         f"http://{DOMAIN}",

core_backend/app/languages/__init__.py

@@ -0,0 +1,3 @@
+from .routers import router
+
+__all__ = ["router"]

core_backend/app/languages/models.py

@@ -0,0 +1,153 @@
+from datetime import datetime
+from datetime import timezone as tz
+from typing import List, Optional
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    Integer,
+    String,
+    delete,
+    select,
+)
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.schema import Index
+
+from ..models import Base
+from .schemas import LanguageBase
+
+
+class LanguageDB(Base):
+    """
+    SQL Alchemy data model for language
+    """
+
+    __tablename__ = "languages"
+
+    language_id: Mapped[int] = mapped_column(Integer, primary_key=True, nullable=False)
+    language_name: Mapped[str] = mapped_column(String, nullable=False)
+    is_default: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    created_datetime_utc: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+    updated_datetime_utc: Mapped[datetime] = mapped_column(DateTime, nullable=False)
+
+    __table_args__ = (
+        Index(
+            "ix_languages_is_default_true",
+            "is_default",
+            unique=True,
+            postgresql_where=(is_default.is_(True)),
+        ),
+    )
+
+    def __repr__(self) -> str:
+        """Pretty Print"""
+        return f"<Language #{self.language_id}:  {self.language_name}>"
+
+
+async def save_language_to_db(
+    language: LanguageBase,
+    asession: AsyncSession,
+) -> LanguageDB:
+    """
+    Saves a new language in the database
+    """
+
+    language_db = LanguageDB(
+        language_name=language.language_name,
+        is_default=language.is_default,
+        created_datetime_utc=datetime.now(tz.utc).replace(tzinfo=None),
+        updated_datetime_utc=datetime.now(tz.utc).replace(tzinfo=None),
+    )
+    asession.add(language_db)
+
+    await asession.commit()
+    await asession.refresh(language_db)
+
+    return language_db
+
+
+async def update_language_in_db(
+    language_id: int,
+    language: LanguageBase,
+    asession: AsyncSession,
+) -> LanguageDB:
+    """
+    Updates a language in the database
+    """
+
+    language_db = LanguageDB(
+        language_id=language_id,
+        is_default=language.is_default,
+        language_name=language.language_name,
+        updated_datetime_utc=datetime.now(tz.utc).replace(tzinfo=None),
+    )
+
+    language_db = await asession.merge(language_db)
+    await asession.commit()
+    return language_db
+
+
+async def delete_language_from_db(
+    language_id: int,
+    asession: AsyncSession,
+) -> None:
+    """
+    Deletes a content from the database
+    """
+    stmt = delete(LanguageDB).where(LanguageDB.language_id == language_id)
+    await asession.execute(stmt)
+    await asession.commit()
+
+
+async def get_language_from_db(
+    language_id: int,
+    asession: AsyncSession,
+) -> Optional[LanguageDB]:
+    """
+    Retrieves a content from the database
+    """
+    stmt = select(LanguageDB).where(LanguageDB.language_id == language_id)
+    language_row = (await asession.execute(stmt)).scalar_one_or_none()
+    return language_row
+
+
+async def get_default_language_from_db(
+    asession: AsyncSession,
+) -> Optional[LanguageDB]:
+    """
+    Retrieves a content from the database
+    """
+    truth_bool = True
+    stmt = select(LanguageDB).where(LanguageDB.is_default == truth_bool)
+    language_row = (await asession.execute(stmt)).scalar_one_or_none()
+    return language_row
+
+
+async def get_list_of_languages_from_db(
+    asession: AsyncSession, offset: int = 0, limit: Optional[int] = None
+) -> List[LanguageDB]:
+    """
+    Retrieves all content from the database
+    """
+    stmt = select(LanguageDB)
+    if offset > 0:
+        stmt = stmt.offset(offset)
+    if limit is not None:
+        stmt = stmt.limit(limit)
+
+    language_rows = (await asession.execute(stmt)).all()
+
+    return [c[0] for c in language_rows] if language_rows else []
+
+
+async def is_language_name_unique(language_name: str, asession: AsyncSession) -> bool:
+    """
+    Check if the language name is unique
+    """
+    stmt = select(LanguageDB).where(LanguageDB.language_name == language_name)
+    language_row = (await asession.execute(stmt)).scalar_one_or_none()
+    if language_row:
+        return False
+    else:
+        return True

core_backend/app/languages/routers.py

@@ -0,0 +1,190 @@
+from typing import Annotated, List
+
+from fastapi import APIRouter, Depends
+from fastapi.exceptions import HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..auth.dependencies import get_current_fullaccess_user, get_current_readonly_user
+from ..auth.schemas import AuthenticatedUser
+from ..database import get_async_session
+from ..utils import setup_logger
+from .models import (
+    LanguageDB,
+    delete_language_from_db,
+    get_default_language_from_db,
+    get_language_from_db,
+    get_list_of_languages_from_db,
+    is_language_name_unique,
+    save_language_to_db,
+    update_language_in_db,
+)
+from .schemas import LanguageBase, LanguageRetrieve
+
+router = APIRouter(prefix="/languages")
+logger = setup_logger()
+
+
+@router.post("/", response_model=LanguageRetrieve)
+async def create_language(
+    language: LanguageBase,
+    full_access_user: Annotated[
+        AuthenticatedUser, Depends(get_current_fullaccess_user)
+    ],
+    asession: AsyncSession = Depends(get_async_session),
+) -> LanguageRetrieve | None:
+    """
+    Create language endpoint. Upsert language to PG database
+    """
+    language.language_name = language.language_name.upper()
+    if not (await is_language_name_unique(language.language_name, asession)):
+        raise HTTPException(status_code=400, detail="Language name already exists")
+
+    if language.is_default is True:
+        await unset_default_language(asession)
+
+    language_db = await save_language_to_db(language, asession)
+
+    return _convert_language_record_to_schema(language_db)
+
+
+@router.put("/{language_id}", response_model=LanguageRetrieve)
+async def edit_language(
+    language_id: int,
+    language: LanguageBase,
+    full_access_user: Annotated[
+        AuthenticatedUser, Depends(get_current_fullaccess_user)
+    ],
+    asession: AsyncSession = Depends(get_async_session),
+) -> LanguageRetrieve | None:
+    """
+    Edit language endpoint.
+    """
+    old_language = await get_language_from_db(
+        language_id,
+        asession,
+    )
+
+    if not old_language:
+        raise HTTPException(
+            status_code=404, detail=f"Language id `{language_id}` not found"
+        )
+    if old_language.is_default and not language.is_default:
+        raise HTTPException(
+            status_code=400,
+            detail="Default language cannot be unset."
+            "Please either create a default language"
+            "or set an existing language as default.",
+        )
+
+    language.language_name = language.language_name.upper()
+    if (old_language.language_name != language.language_name) and (
+        not (await is_language_name_unique(language.language_name, asession))
+    ):
+        raise HTTPException(status_code=400, detail="Language name already exists")
+
+    if language.is_default and not old_language.is_default:
+        await unset_default_language(asession)
+
+    updated_language = await update_language_in_db(
+        language_id,
+        language,
+        asession,
+    )
+
+    return _convert_language_record_to_schema(updated_language)
+
+
+@router.get("/", response_model=list[LanguageRetrieve])
+async def retrieve_languages(
+    readonly_user: Annotated[AuthenticatedUser, Depends(get_current_readonly_user)],
+    asession: AsyncSession = Depends(get_async_session),
+) -> List[LanguageRetrieve]:
+    """
+    Retrieve languages endpoint
+    """
+    languages = await get_list_of_languages_from_db(asession)
+    return [_convert_language_record_to_schema(language) for language in languages]
+
+
+@router.get("/default", response_model=LanguageRetrieve)
+async def retrieve_default_language(
+    readonly_user: Annotated[AuthenticatedUser, Depends(get_current_readonly_user)],
+    asession: AsyncSession = Depends(get_async_session),
+) -> LanguageRetrieve:
+    """
+    Retrieve default language endpoint
+    """
+    language = await get_default_language_from_db(asession)
+    if not language:
+        raise HTTPException(status_code=404, detail="Default language not found")
+    return _convert_language_record_to_schema(language)
+
+
+@router.get("/{language_id}", response_model=LanguageRetrieve)
+async def retrieve_language_by_id(
+    language_id: int,
+    readonly_user: Annotated[AuthenticatedUser, Depends(get_current_readonly_user)],
+    asession: AsyncSession = Depends(get_async_session),
+) -> LanguageRetrieve:
+    """
+    Retrieve language by id endpoint
+    """
+    language = await get_language_from_db(language_id, asession)
+    if not language:
+        raise HTTPException(
+            status_code=404, detail=f"Language id `{language_id}` not found"
+        )
+    return _convert_language_record_to_schema(language)
+
+
+@router.delete("/{language_id}")
+async def delete_language(
+    language_id: int,
+    full_access_user: Annotated[
+        AuthenticatedUser, Depends(get_current_fullaccess_user)
+    ],
+    asession: AsyncSession = Depends(get_async_session),
+) -> None:
+    """
+    Delete language endpoint
+    """
+    language = await get_language_from_db(
+        language_id,
+        asession,
+    )
+
+    if not language:
+        raise HTTPException(
+            status_code=404, detail=f"Language id `{language_id}` not found"
+        )
+
+    if language.is_default:
+        raise HTTPException(
+            status_code=400, detail="Default language cannot be deleted"
+        )
+    await delete_language_from_db(language_id, asession)
+
+
+async def unset_default_language(asession: AsyncSession) -> None:
+    """
+    Unset default language
+    """
+    default_language = await get_default_language_from_db(asession)
+
+    if default_language:
+        default_language.is_default = False
+        default_language_schema = _convert_language_record_to_schema(default_language)
+        await update_language_in_db(
+            default_language.language_id, default_language_schema, asession
+        )
+    return None
+
+
+def _convert_language_record_to_schema(language_db: LanguageDB) -> LanguageRetrieve:
+    return LanguageRetrieve(
+        language_id=language_db.language_id,
+        language_name=language_db.language_name,
+        is_default=language_db.is_default,
+        created_datetime_utc=language_db.created_datetime_utc,
+        updated_datetime_utc=language_db.updated_datetime_utc,
+    )