refactor: organization routes/crud/schemas to use best practice (#1096)

* fix: use enums for HTTPStatus over codes

* refactor: update organization routes/crud/schemas best pracice

* added int to the union with str in org_exists functions

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: Niraj Adhikari <[email protected]>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

src/backend/app/models/enums.py

@@ -27,6 +27,28 @@ class IntEnum(int, Enum):
     pass
 
 
+class HTTPStatus(IntEnum):
+    """All HTTP status codes used in endpoints."""
+
+    # Success
+    OK = 200
+    CREATED = 201
+    ACCEPTED = 202
+    NO_CONTENT = 204
+
+    # Client Error
+    BAD_REQUEST = 400
+    UNAUTHORIZED = 401
+    FORBIDDEN = 403
+    NOT_FOUND = 404
+    CONFLICT = 409
+    UNPROCESSABLE_ENTITY = 422
+
+    # Server Error
+    INTERNAL_SERVER_ERROR = 500
+    NOT_IMPLEMENTED = 501
+
+
 class TeamVisibility(IntEnum, Enum):
     """Describes the visibility associated with an Team."""
 

src/backend/app/organization/organization_crud.py

@@ -17,48 +17,28 @@
 #
 """Logic for organization management."""
 
-import re
 from io import BytesIO
 
-from fastapi import HTTPException, UploadFile
+from fastapi import HTTPException, Response, UploadFile
 from loguru import logger as log
-from sqlalchemy import func
+from sqlalchemy import update
 from sqlalchemy.orm import Session
 
 from app.config import settings
 from app.db import db_models
+from app.models.enums import HTTPStatus
+from app.organization.organization_deps import (
+    get_organization_by_name,
+)
+from app.organization.organization_schemas import OrganisationEdit, OrganisationIn
 from app.s3 import add_obj_to_bucket
 
 
 def get_organisations(
     db: Session,
 ):
     """Get all orgs."""
-    db_organisation = db.query(db_models.DbOrganisation).all()
-    return db_organisation
-
-
-def generate_slug(text: str) -> str:
-    """Sanitise the organization name for use in a URL."""
-    # Remove special characters and replace spaces with hyphens
-    slug = re.sub(r"[^\w\s-]", "", text).strip().lower().replace(" ", "-")
-    # Remove consecutive hyphens
-    slug = re.sub(r"[-\s]+", "-", slug)
-    return slug
-
-
-async def get_organisation_by_name(db: Session, name: str):
-    """Get org by name.
-
-    This function is used to check if a org exists with the same name.
-    """
-    # Use SQLAlchemy's query-building capabilities
-    db_organisation = (
-        db.query(db_models.DbOrganisation)
-        .filter(func.lower(db_models.DbOrganisation.name).like(func.lower(f"%{name}%")))
-        .first()
-    )
-    return db_organisation
+    return db.query(db_models.DbOrganisation).all()
 
 
 async def upload_logo_to_s3(
@@ -94,32 +74,33 @@ async def upload_logo_to_s3(
 
 
 async def create_organization(
-    db: Session, name: str, description: str, url: str, logo: UploadFile(None)
-):
+    db: Session, org_model: OrganisationIn, logo: UploadFile(None)
+) -> db_models.DbOrganisation:
     """Creates a new organization with the given name, description, url, type, and logo.
 
     Saves the logo file S3 bucket under /{org_id}/logo.png.
 
     Args:
         db (Session): database session
-        name (str): name of the organization
-        description (str): description of the organization
-        url (str): url of the organization
-        type (int): type of the organization
+        org_model (OrganisationIn): Pydantic model for organization input.
         logo (UploadFile, optional): logo file of the organization.
             Defaults to File(...).
 
     Returns:
-        bool: True if organization was created successfully
+        DbOrganization: SQLAlchemy Organization model.
     """
+    if await get_organization_by_name(db, org_name=org_model.name):
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT,
+            detail=f"Organization already exists with the name {org_model.name}",
+        )
+
+    # Required to check if exists on error
+    db_organization = None
+
     try:
         # Create new organization without logo set
-        db_organization = db_models.DbOrganisation(
-            name=name,
-            slug=generate_slug(name),
-            description=description,
-            url=url,
-        )
+        db_organization = db_models.DbOrganisation(**org_model.dict())
 
         db.add(db_organization)
         db.commit()
@@ -145,49 +126,63 @@ async def create_organization(
             status_code=400, detail=f"Error creating organization: {e}"
         ) from e
 
-    return True
+    return db_organization
 
 
-async def get_organisation_by_id(db: Session, id: int):
-    """Get an organization by its id.
+async def update_organization(
+    db: Session,
+    organization: db_models.DbOrganisation,
+    values: OrganisationEdit,
+    logo: UploadFile(None),
+) -> db_models.DbOrganisation:
+    """Update an existing organisation database entry.
 
     Args:
         db (Session): database session
-        id (int): id of the organization
+        organization (DbOrganisation): Editing database model.
+        values (OrganisationEdit): Pydantic model for organization edit.
+        logo (UploadFile, optional): logo file of the organization.
+            Defaults to File(...).
 
     Returns:
-        DbOrganisation: organization with the given id
+        DbOrganization: SQLAlchemy Organization model.
     """
-    db_organization = (
-        db.query(db_models.DbOrganisation)
-        .filter(db_models.DbOrganisation.id == id)
-        .first()
-    )
-    return db_organization
+    if not (updated_fields := values.dict(exclude_none=True)):
+        raise HTTPException(
+            status_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+            detail=f"No values were provided to update organization {organization.id}",
+        )
 
+    update_cmd = (
+        update(db_models.DbOrganisation)
+        .where(db_models.DbOrganisation.id == organization.id)
+        .values(**updated_fields)
+    )
+    db.execute(update_cmd)
 
-async def update_organization_info(
-    db: Session,
-    organization_id,
-    name: str,
-    description: str,
-    url: str,
-    logo: UploadFile,
-):
-    """Update an existing organisation database entry."""
-    organization = await get_organisation_by_id(db, organization_id)
-    if not organization:
-        raise HTTPException(status_code=404, detail="Organization not found")
-
-    if name:
-        organization.name = name
-    if description:
-        organization.description = description
-    if url:
-        organization.url = url
     if logo:
         organization.logo = await upload_logo_to_s3(organization, logo)
 
     db.commit()
     db.refresh(organization)
+
     return organization
+
+
+async def delete_organization(
+    db: Session,
+    organization: db_models.DbOrganisation,
+) -> Response:
+    """Delete an existing organisation database entry.
+
+    Args:
+        db (Session): database session
+        organization (DbOrganisation): Database model to delete.
+
+    Returns:
+        bool: If deletion was successful.
+    """
+    db.delete(organization)
+    db.commit()
+
+    return Response(status_code=HTTPStatus.NO_CONTENT)

src/backend/app/organization/organization_deps.py

@@ -0,0 +1,92 @@
+# Copyright (c) 2022, 2023 Humanitarian OpenStreetMap Team
+#
+# This file is part of FMTM.
+#
+#     FMTM is free software: you can redistribute it and/or modify
+#     it under the terms of the GNU General Public License as published by
+#     the Free Software Foundation, either version 3 of the License, or
+#     (at your option) any later version.
+#
+#     FMTM is distributed in the hope that it will be useful,
+#     but WITHOUT ANY WARRANTY; without even the implied warranty of
+#     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#     GNU General Public License for more details.
+#
+#     You should have received a copy of the GNU General Public License
+#     along with FMTM.  If not, see <https:#www.gnu.org/licenses/>.
+#
+
+"""Organization dependencies for use in Depends."""
+
+from typing import Union
+
+from fastapi import Depends
+from fastapi.exceptions import HTTPException
+from loguru import logger as log
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.db.database import get_db
+from app.db.db_models import DbOrganisation
+from app.models.enums import HTTPStatus
+
+
+async def get_organization_by_name(db: Session, org_name: str) -> DbOrganisation:
+    """Get an organization from the db by name.
+
+    Args:
+        db (Session): database session
+        org_name (int): id of the organization
+
+    Returns:
+        DbOrganisation: organization with the given id
+    """
+    return (
+        db.query(DbOrganisation)
+        .filter(func.lower(DbOrganisation.name).like(func.lower(f"%{org_name}%")))
+        .first()
+    )
+
+
+async def get_organisation_by_id(db: Session, org_id: int) -> DbOrganisation:
+    """Get an organization from the db by id.
+
+    Args:
+        db (Session): database session
+        org_id (int): id of the organization
+
+    Returns:
+        DbOrganisation: organization with the given id
+    """
+    return db.query(DbOrganisation).filter(DbOrganisation.id == org_id).first()
+
+
+async def org_exists(
+    org_id: Union[str, int],
+    db: Session = Depends(get_db),
+) -> DbOrganisation:
+    """Check if organization name exists, else error.
+
+    The org_id can also be an org name.
+    """
+    try:
+        org_id = int(org_id)
+    except ValueError:
+        pass
+
+    if isinstance(org_id, int):
+        log.debug(f"Getting organization by id: {org_id}")
+        db_organization = await get_organisation_by_id(db, org_id)
+
+    if isinstance(org_id, str):
+        log.debug(f"Getting organization by name: {org_id}")
+        db_organization = await get_organization_by_name(db, org_id)
+
+    if not db_organization:
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail=f"Organization {org_id} does not exist",
+        )
+
+    log.debug(f"Organization match: {db_organization}")
+    return db_organization