Recommendation

app/modules/recommendation/cruds_recommendation.py

@@ -0,0 +1,70 @@
+from collections.abc import Sequence
+
+from sqlalchemy import delete, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.modules.recommendation import models_recommendation, schemas_recommendation
+
+
+async def get_recommendations(
+    db: AsyncSession,
+) -> Sequence[models_recommendation.Recommendation]:
+    result = await db.execute(select(models_recommendation.Recommendation))
+    return result.scalars().all()
+
+
+async def create_recommendation(
+    recommendation: models_recommendation.Recommendation,
+    db: AsyncSession,
+) -> models_recommendation.Recommendation:
+    db.add(recommendation)
+    await db.commit()
+    return recommendation
+
+
+async def update_recommendation(
+    recommendation_id: str,
+    recommendation: schemas_recommendation.RecommendationEdit,
+    db: AsyncSession,
+):
+    if not any(recommendation.model_dump().values()):
+        return
+
+    result = await db.execute(
+        update(models_recommendation.Recommendation)
+        .where(models_recommendation.Recommendation.id == recommendation_id)
+        .values(**recommendation.model_dump(exclude_none=True)),
+    )
+    if result.rowcount == 1:
+        await db.commit()
+    else:
+        await db.rollback()
+        raise ValueError
+
+
+async def delete_recommendation(
+    recommendation_id: str,
+    db: AsyncSession,
+):
+    result = await db.execute(
+        delete(models_recommendation.Recommendation).where(
+            models_recommendation.Recommendation.id == recommendation_id,
+        ),
+    )
+    if result.rowcount == 1:
+        await db.commit()
+    else:
+        await db.rollback()
+        raise ValueError
+
+
+async def get_recommendation_by_id(
+    recommendation_id: str,
+    db: AsyncSession,
+) -> models_recommendation.Recommendation | None:
+    result = await db.execute(
+        select(models_recommendation.Recommendation).where(
+            models_recommendation.Recommendation.id == recommendation_id,
+        ),
+    )
+    return result.scalars().one_or_none()

app/modules/recommendation/endpoints_recommendation.py

@@ -0,0 +1,197 @@
+import uuid
+from datetime import UTC, datetime
+
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from fastapi.responses import FileResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core import models_core, standard_responses
+from app.core.config import Settings
+from app.core.groups.groups_type import GroupType
+from app.core.module import Module
+from app.dependencies import (
+    get_db,
+    get_request_id,
+    get_settings,
+    is_user_a_member,
+    is_user_a_member_of,
+)
+from app.modules.recommendation import (
+    cruds_recommendation,
+    models_recommendation,
+    schemas_recommendation,
+)
+from app.utils.tools import get_file_from_data, save_file_as_data
+
+router = APIRouter()
+
+
+module = Module(
+    root="recommendation",
+    tag="Recommendation",
+    default_allowed_groups_ids=[GroupType.student, GroupType.staff],
+)
+
+
+@module.router.get(
+    "/recommendation/recommendations",
+    response_model=list[schemas_recommendation.Recommendation],
+    status_code=200,
+)
+async def get_recommendation(
+    db: AsyncSession = Depends(get_db),
+    user: models_core.CoreUser = Depends(is_user_a_member),
+):
+    """
+    Get recommendations.
+
+    **The user must be authenticated to use this endpoint**
+    """
+
+    return await cruds_recommendation.get_recommendations(db=db)
+
+
+@module.router.post(
+    "/recommendation/recommendations",
+    response_model=schemas_recommendation.Recommendation,
+    status_code=201,
+)
+async def create_recommendation(
+    recommendation: schemas_recommendation.RecommendationBase,
+    db: AsyncSession = Depends(get_db),
+    settings: Settings = Depends(get_settings),
+    user: models_core.CoreUser = Depends(is_user_a_member_of(GroupType.BDE)),
+):
+    """
+    Create a recommendation.
+
+    **This endpoint is only usable by members of the group BDE**
+    """
+
+    recommendation_db = models_recommendation.Recommendation(
+        id=str(uuid.uuid4()),
+        creation=datetime.now(UTC),
+        **recommendation.model_dump(),
+    )
+
+    return await cruds_recommendation.create_recommendation(
+        recommendation=recommendation_db,
+        db=db,
+    )
+
+
+@module.router.patch(
+    "/recommendation/recommendations/{recommendation_id}",
+    status_code=204,
+)
+async def edit_recommendation(
+    recommendation_id: str,
+    recommendation: schemas_recommendation.RecommendationEdit,
+    db: AsyncSession = Depends(get_db),
+    user: models_core.CoreUser = Depends(is_user_a_member_of(GroupType.BDE)),
+):
+    """
+    Edit a recommendation.
+
+    **This endpoint is only usable by members of the group BDE**
+    """
+
+    try:
+        await cruds_recommendation.update_recommendation(
+            recommendation_id=recommendation_id,
+            recommendation=recommendation,
+            db=db,
+        )
+    except ValueError:
+        raise HTTPException(status_code=404, detail="The recommendation does not exist")
+
+
+@module.router.delete(
+    "/recommendation/recommendations/{recommendation_id}",
+    status_code=204,
+)
+async def delete_recommendation(
+    recommendation_id: str,
+    db: AsyncSession = Depends(get_db),
+    user: models_core.CoreUser = Depends(is_user_a_member_of(GroupType.BDE)),
+):
+    """
+    Delete a recommendation.
+
+    **This endpoint is only usable by members of the group BDE**
+    """
+
+    try:
+        await cruds_recommendation.delete_recommendation(
+            db=db,
+            recommendation_id=recommendation_id,
+        )
+    except ValueError:
+        raise HTTPException(status_code=404, detail="The recommendation does not exist")
+
+
+@module.router.get(
+    "/recommendation/recommendations/{recommendation_id}/picture",
+    response_class=FileResponse,
+    status_code=200,
+)
+async def read_recommendation_image(
+    recommendation_id: str,
+    db: AsyncSession = Depends(get_db),
+    user: models_core.CoreUser = Depends(is_user_a_member),
+):
+    """
+    Get the image of a recommendation.
+
+    **The user must be authenticated to use this endpoint**
+    """
+    recommendation = await cruds_recommendation.get_recommendation_by_id(
+        recommendation_id=recommendation_id,
+        db=db,
+    )
+
+    if not recommendation:
+        raise HTTPException(status_code=404, detail="The recommendation does not exist")
+
+    return get_file_from_data(
+        default_asset="assets/images/default_recommendation.png",
+        directory="recommendations",
+        filename=recommendation_id,
+    )
+
+
+@module.router.post(
+    "/recommendation/recommendations/{recommendation_id}/picture",
+    response_model=standard_responses.Result,
+    status_code=201,
+)
+async def create_recommendation_image(
+    recommendation_id: str,
+    image: UploadFile = File(),
+    user: models_core.CoreUser = Depends(is_user_a_member_of(GroupType.BDE)),
+    request_id: str = Depends(get_request_id),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Add an image to a recommendation.
+
+    **This endpoint is only usable by members of the group BDE**
+    """
+    recommendation = await cruds_recommendation.get_recommendation_by_id(
+        recommendation_id=recommendation_id,
+        db=db,
+    )
+
+    if not recommendation:
+        raise HTTPException(status_code=404, detail="The recommendation does not exist")
+
+    await save_file_as_data(
+        image=image,
+        directory="recommendations",
+        filename=str(recommendation_id),
+        request_id=request_id,
+        max_file_size=4 * 1024 * 1024,
+        accepted_content_types=["image/jpeg", "image/png", "image/webp"],
+    )
+
+    return standard_responses.Result(success=True)

app/modules/recommendation/models_recommendation.py

@@ -0,0 +1,19 @@
+from datetime import datetime
+
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+from app.utils.types.datetime import TZDateTime
+
+
+class Recommendation(Base):
+    __tablename__ = "recommendation"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True, nullable=False)
+    creation: Mapped[datetime] = mapped_column(TZDateTime, nullable=False)
+    title: Mapped[str] = mapped_column(String, nullable=False)
+    code: Mapped[str | None] = mapped_column(String, nullable=True)
+
+    summary: Mapped[str] = mapped_column(String, nullable=False)
+    description: Mapped[str] = mapped_column(String, nullable=False)

app/modules/recommendation/schemas_recommendation.py

@@ -0,0 +1,24 @@
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class RecommendationBase(BaseModel):
+    title: str
+    code: str | None = None
+    summary: str
+    description: str
+
+
+class Recommendation(RecommendationBase):
+    id: str
+    creation: datetime
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class RecommendationEdit(BaseModel):
+    title: str | None = None
+    code: str | None = None
+    summary: str | None = None
+    description: str | None = None

migrations/versions/4-recommendation.py

@@ -0,0 +1,39 @@
+"""add migration file for recommendation module
+
+Revision ID: 3f9843f165e9
+Revises: 99a2c70e4a24
+Create Date: 2024-03-16 06:02:34.664485
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "3f9843f165e9"
+down_revision: str | None = "99a2c70e4a24"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "recommendation",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("creation", sa.DateTime(timezone=False), nullable=False),
+        sa.Column("title", sa.String(), nullable=False),
+        sa.Column("code", sa.String(), nullable=True),
+        sa.Column("summary", sa.String(), nullable=False),
+        sa.Column("description", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("recommendation")
+    # ### end Alembic commands ###