CoreAssociation

app/core/associations/cruds_associations.py

@@ -0,0 +1,70 @@
+import uuid
+from collections.abc import Sequence
+from uuid import UUID
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.associations import models_associations, schemas_associations
+
+
+async def get_associations(
+    db: AsyncSession,
+) -> Sequence[models_associations.CoreAssociation]:
+    result = await db.execute(select(models_associations.CoreAssociation))
+    return result.scalars().all()
+
+
+async def get_associations_for_groups(
+    group_ids: list[str],
+    db: AsyncSession,
+) -> Sequence[models_associations.CoreAssociation]:
+    result = await db.execute(
+        select(models_associations.CoreAssociation).where(
+            models_associations.CoreAssociation.group_id.in_(group_ids),
+        ),
+    )
+    return result.scalars().all()
+
+
+async def get_association_by_id(
+    association_id: uuid.UUID,
+    db: AsyncSession,
+) -> models_associations.CoreAssociation | None:
+    result = await db.execute(
+        select(models_associations.CoreAssociation).where(
+            models_associations.CoreAssociation.id == association_id,
+        ),
+    )
+    return result.scalars().first()
+
+
+async def get_association_by_name(
+    name: str,
+    db: AsyncSession,
+) -> models_associations.CoreAssociation | None:
+    result = await db.execute(
+        select(models_associations.CoreAssociation).where(
+            models_associations.CoreAssociation.name == name,
+        ),
+    )
+    return result.scalars().first()
+
+
+async def create_association(
+    db: AsyncSession,
+    association: models_associations.CoreAssociation,
+) -> None:
+    db.add(association)
+
+
+async def update_association(
+    db: AsyncSession,
+    association_id: UUID,
+    association_update: schemas_associations.AssociationUpdate,
+) -> None:
+    await db.execute(
+        update(models_associations.CoreAssociation)
+        .where(models_associations.CoreAssociation.id == association_id)
+        .values(**association_update.model_dump(exclude_unset=True)),
+    )

app/core/associations/endpoints_associations.py

@@ -0,0 +1,216 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, UploadFile
+from fastapi.responses import FileResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.associations import (
+    cruds_associations,
+    models_associations,
+    schemas_associations,
+)
+from app.core.associations.factory_associations import AssociationsFactory
+from app.core.groups.groups_type import GroupType
+from app.core.users import models_users
+from app.dependencies import (
+    get_db,
+    is_user,
+    is_user_in,
+)
+from app.types.content_type import ContentType
+from app.types.module import CoreModule
+from app.utils.tools import get_file_from_data, save_file_as_data
+
+router = APIRouter(tags=["Associations"])
+
+core_module = CoreModule(
+    root="associations",
+    tag="Associations",
+    router=router,
+    factory=AssociationsFactory(),
+)
+
+
+@router.get(
+    "/associations/",
+    response_model=list[schemas_associations.Association],
+    status_code=200,
+)
+async def read_associations(
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(is_user()),
+):
+    """
+    Return all associations
+
+    **User must be authenticated**
+    """
+
+    return await cruds_associations.get_associations(db=db)
+
+
+@router.get(
+    "/associations/me",
+    response_model=list[schemas_associations.Association],
+    status_code=200,
+)
+async def read_associations_me(
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(is_user()),
+):
+    """
+    Return all associations the current user has the right to manage
+
+    **User must be authenticated**
+    """
+
+    return await cruds_associations.get_associations_for_groups(
+        group_ids=user.group_ids,
+        db=db,
+    )
+
+
+@router.post(
+    "/associations/",
+    response_model=schemas_associations.Association,
+    status_code=201,
+)
+async def create_association(
+    association: schemas_associations.AssociationBase,
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(is_user_in(GroupType.admin)),
+):
+    """
+    Create a new association
+
+    **This endpoint is only usable by administrators**
+    """
+    if (
+        await cruds_associations.get_association_by_name(name=association.name, db=db)
+        is not None
+    ):
+        raise HTTPException(
+            status_code=400,
+            detail="A association with this name already exist",
+        )
+
+    db_association = models_associations.CoreAssociation(
+        id=uuid.uuid4(),
+        name=association.name,
+        group_id=association.group_id,
+    )
+    await cruds_associations.create_association(
+        association=db_association,
+        db=db,
+    )
+
+    return db_association
+
+
+@router.patch(
+    "/associations/{association_id}",
+    status_code=204,
+)
+async def update_association(
+    association_id: uuid.UUID,
+    association_update: schemas_associations.AssociationUpdate,
+    db: AsyncSession = Depends(get_db),
+    user=Depends(is_user_in(GroupType.admin)),
+):
+    """
+    Update the name or the description of a association.
+
+    **This endpoint is only usable by administrators**
+    """
+    association = await cruds_associations.get_association_by_id(
+        association_id=association_id,
+        db=db,
+    )
+    if not association:
+        raise HTTPException(status_code=404, detail="Association not found")
+
+    # If the request ask to update the association name, we need to check it is available
+    if association_update.name and association_update.name != association.name:
+        if (
+            await cruds_associations.get_association_by_name(
+                name=association_update.name,
+                db=db,
+            )
+            is not None
+        ):
+            raise HTTPException(
+                status_code=400,
+                detail="A association with the name already exist",
+            )
+
+    await cruds_associations.update_association(
+        db=db,
+        association_id=association_id,
+        association_update=association_update,
+    )
+
+
+@router.post(
+    "/associations/{association_id}/logo",
+    status_code=204,
+)
+async def create_association_logo(
+    association_id: uuid.UUID,
+    image: UploadFile,
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(is_user_in(GroupType.admin)),
+):
+    """
+    Upload a logo for an association
+
+    **This endpoint is only usable by administrators**
+    """
+
+    association = await cruds_associations.get_association_by_id(
+        db=db,
+        association_id=association_id,
+    )
+    if not association:
+        raise HTTPException(status_code=404, detail="Association not found")
+
+    await save_file_as_data(
+        upload_file=image,
+        directory="associations/logos",
+        filename=association_id,
+        max_file_size=4 * 1024 * 1024,
+        accepted_content_types=[
+            ContentType.jpg,
+            ContentType.png,
+            ContentType.webp,
+        ],
+    )
+
+
+@router.get(
+    "/associations/{association_id}/logo",
+    response_class=FileResponse,
+    status_code=200,
+)
+async def read_association_logo(
+    association_id: uuid.UUID,
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(is_user()),
+):
+    """
+    Get the logo of an association
+
+    **User must be authenticated**
+    """
+
+    association = await cruds_associations.get_association_by_id(
+        db=db,
+        association_id=association_id,
+    )
+    if not association:
+        raise HTTPException(status_code=404, detail="Association not found")
+
+    return get_file_from_data(
+        directory="associations/logos",
+        filename=association_id,
+        raise_http_exception=True,
+    )

app/core/associations/factory_associations.py

@@ -0,0 +1,39 @@
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.associations import cruds_associations
+from app.core.associations.models_associations import CoreAssociation
+from app.core.groups.factory_groups import CoreGroupsFactory
+from app.core.utils.config import Settings
+from app.types.factory import Factory
+
+
+class AssociationsFactory(Factory):
+    association_ids = [
+        uuid.uuid4(),
+        uuid.uuid4(),
+    ]
+
+    depends_on = [CoreGroupsFactory]
+
+    @classmethod
+    async def create_associations(cls, db: AsyncSession):
+        descriptions = ["Association 1", "Association 2"]
+        for i in range(len(CoreGroupsFactory.groups_ids)):
+            await cruds_associations.create_association(
+                db=db,
+                association=CoreAssociation(
+                    id=cls.association_ids[i],
+                    name=descriptions[i],
+                    group_id=CoreGroupsFactory.groups_ids[i],
+                ),
+            )
+
+    @classmethod
+    async def run(cls, db: AsyncSession, settings: Settings) -> None:
+        await cls.create_associations(db=db)
+
+    @classmethod
+    async def should_run(cls, db: AsyncSession):
+        return len(await cruds_associations.get_associations(db=db)) > 0

app/core/associations/models_associations.py

@@ -0,0 +1,12 @@
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.types.sqlalchemy import Base, PrimaryKey
+
+
+class CoreAssociation(Base):
+    __tablename__ = "associations_associations"
+
+    id: Mapped[PrimaryKey]
+    name: Mapped[str] = mapped_column(unique=True)
+    group_id: Mapped[str] = mapped_column(ForeignKey("core_group.id"))

app/core/associations/schemas_associations.py

@@ -0,0 +1,17 @@
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class AssociationBase(BaseModel):
+    name: str
+    group_id: str
+
+
+class Association(AssociationBase):
+    id: UUID
+
+
+class AssociationUpdate(BaseModel):
+    name: str | None = None
+    group_id: str | None = None

app/core/users/models_users.py

@@ -63,6 +63,10 @@ def full_name(self) -> str:
             return f"{self.firstname} {self.name} ({self.nickname})"
         return f"{self.firstname} {self.name}"
 
+    @property
+    def group_ids(self) -> list[str]:
+        return [group.id for group in self.groups]
+
 
 class CoreUserUnconfirmed(Base):
     __tablename__ = "core_user_unconfirmed"