Organization/project : Crud, Endpoint and Test Cases

backend/app/api/main.py

@@ -1,13 +1,17 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, private, users, utils
+
+from app.api.routes import items, login, private, users, utils,project,organization
+
 from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
 api_router.include_router(users.router)
 api_router.include_router(utils.router)
 api_router.include_router(items.router)
+api_router.include_router(organization.router)
+api_router.include_router(project.router)
 
 
 if settings.ENVIRONMENT == "local":

backend/app/api/routes/organization.py

@@ -0,0 +1,76 @@
+from typing import Any, List
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import func
+from sqlmodel import Session, select
+
+from app.models import Organization, OrganizationCreate, OrganizationUpdate, OrganizationPublic
+from app.api.deps import (
+    CurrentUser,
+    SessionDep,
+    get_current_active_superuser,
+)
+from app.crud.organization import create_organization, get_organization_by_id
+from app.utils import APIResponse 
+
+router = APIRouter(prefix="/organizations", tags=["organizations"])
+
+
+# Retrieve organizations
+@router.get("/", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[List[OrganizationPublic]])
+def read_organizations(session: SessionDep, skip: int = 0, limit: int = 100):
+    count_statement = select(func.count()).select_from(Organization)
+    count = session.exec(count_statement).one()
+
+    statement = select(Organization).offset(skip).limit(limit)
+    organizations = session.exec(statement).all()
+
+    return APIResponse.success_response(organizations)
+
+# Create a new organization
+@router.post("/", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[OrganizationPublic])
+def create_new_organization(*, session: SessionDep, org_in: OrganizationCreate):
+    new_org = create_organization(session=session, org_create=org_in)
+    return APIResponse.success_response(new_org)
+
+
+@router.get("/{org_id}", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[OrganizationPublic])
+def read_organization(*, session: SessionDep, org_id: int):
+    """
+    Retrieve an organization by ID.
+    """
+    org = get_organization_by_id(session=session, org_id=org_id)
+    if org is None:
+        raise HTTPException(status_code=404, detail="Organization not found")
+    return APIResponse.success_response(org)
+
+
+# Update an organization
+@router.patch("/{org_id}", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[OrganizationPublic])
+def update_organization(*, session: SessionDep, org_id: int, org_in: OrganizationUpdate):
+    org = get_organization_by_id(session=session, org_id=org_id)
+    if org is None:
+        raise HTTPException(status_code=404, detail="Organization not found")
+
+    org_data = org_in.model_dump(exclude_unset=True)   
+    org = org.model_copy(update=org_data)
+
+
+    session.add(org)
+    session.commit()
+    session.flush()
+
+    return APIResponse.success_response(org)
+
+
+# Delete an organization
+@router.delete("/{org_id}", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[None])
+def delete_organization(session: SessionDep, org_id: int):
+    org = get_organization_by_id(session=session, org_id=org_id)
+    if org is None:
+        raise HTTPException(status_code=404, detail="Organization not found")
+
+    session.delete(org)
+    session.commit()
+
+    return APIResponse.success_response(None)

backend/app/api/routes/project.py

@@ -0,0 +1,74 @@
+from typing import Any, List
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import func
+from sqlmodel import Session, select
+
+from app.models import Project, ProjectCreate, ProjectUpdate, ProjectPublic
+from app.api.deps import (
+    CurrentUser,
+    SessionDep,
+    get_current_active_superuser,
+)
+from app.crud.project import create_project, get_project_by_id, get_projects_by_organization
+from app.utils import APIResponse
+
+router = APIRouter(prefix="/projects", tags=["projects"])
+
+
+# Retrieve projects
+@router.get("/",dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[List[ProjectPublic]])
+def read_projects(session: SessionDep, skip: int = 0, limit: int = 100):
+    count_statement = select(func.count()).select_from(Project)
+    count = session.exec(count_statement).one()
+
+    statement = select(Project).offset(skip).limit(limit)
+    projects = session.exec(statement).all()
+
+    return APIResponse.success_response(projects)
+
+
+# Create a new project
+@router.post("/", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[ProjectPublic])
+def create_new_project(*, session: SessionDep, project_in: ProjectCreate):
+    project = create_project(session=session, project_create=project_in)
+    return APIResponse.success_response(project)
+
+@router.get("/{project_id}", dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[ProjectPublic])
+def read_project(*, session: SessionDep, project_id: int) :
+    """
+    Retrieve a project by ID.
+    """
+    project = get_project_by_id(session=session, project_id=project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="Project not found")
+    return APIResponse.success_response(project)
+
+
+# Update a project
+@router.patch("/{project_id}",dependencies=[Depends(get_current_active_superuser)], response_model=APIResponse[ProjectPublic])
+def update_project(*, session: SessionDep, project_id: int, project_in: ProjectUpdate):
+    project = get_project_by_id(session=session, project_id=project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="Project not found")
+
+    project_data = project_in.model_dump(exclude_unset=True)
+    project = project.model_copy(update=project_data)
+
+    session.add(project)
+    session.commit()
+    session.flush()
+    return APIResponse.success_response(project)
+
+
+# Delete a project
+@router.delete("/{project_id}",dependencies=[Depends(get_current_active_superuser)])
+def delete_project(session: SessionDep, project_id: int):
+    project = get_project_by_id(session=session, project_id=project_id)
+    if project is None:
+        raise HTTPException(status_code=404, detail="Project not found")
+
+    session.delete(project)
+    session.commit()
+
+    return APIResponse.success_response(None)

backend/app/crud/organization.py

@@ -0,0 +1,21 @@
+from typing import Any, Optional
+
+from sqlmodel import Session, select
+
+from app.models import Organization, OrganizationCreate
+
+def create_organization(*, session: Session, org_create: OrganizationCreate) -> Organization:
+    db_org = Organization.model_validate(org_create)
+    session.add(db_org)
+    session.commit()
+    session.refresh(db_org)
+    return db_org
+
+
+def get_organization_by_id(*, session: Session, org_id: int) -> Optional[Organization]:
+    statement = select(Organization).where(Organization.id == org_id)
+    return session.exec(statement).first()
+
+def get_organization_by_name(*, session: Session, name: str) -> Optional[Organization]:
+    statement = select(Organization).where(Organization.name == name)
+    return session.exec(statement).first()

backend/app/crud/project.py

@@ -0,0 +1,21 @@
+from typing import List, Optional
+
+from sqlmodel import Session, select
+
+from app.models import Project, ProjectCreate
+
+
+def create_project(*, session: Session, project_create: ProjectCreate) -> Project:
+    db_project = Project.model_validate(project_create)
+    session.add(db_project)
+    session.commit()
+    session.refresh(db_project)
+    return db_project
+
+def get_project_by_id(*, session: Session, project_id: int) -> Optional[Project]:
+    statement = select(Project).where(Project.id == project_id)
+    return session.exec(statement).first()
+
+def get_projects_by_organization(*, session: Session, org_id: int) -> List[Project]:
+    statement = select(Project).where(Project.organization_id == org_id)
+    return session.exec(statement).all()

backend/app/models/__init__.py

@@ -11,4 +11,18 @@
     UserUpdateMe,
     NewPassword,
     UpdatePassword,
+)
+from .organization import (
+    Organization,
+    OrganizationCreate,
+    OrganizationPublic,
+    OrganizationsPublic,
+    OrganizationUpdate,
+)
+from .project import (
+    Project,
+    ProjectCreate,
+    ProjectPublic,
+    ProjectsPublic,
+    ProjectUpdate,
 )

backend/app/models/organization.py

@@ -0,0 +1,33 @@
+from sqlmodel import Field, Relationship, SQLModel
+
+
+# Shared properties for an Organization
+class OrganizationBase(SQLModel):
+    name: str = Field(unique=True, index=True, max_length=255)
+    is_active: bool = True
+
+
+# Properties to receive via API on creation
+class OrganizationCreate(OrganizationBase):
+    pass
+
+
+# Properties to receive via API on update, all are optional
+class OrganizationUpdate(SQLModel):
+    name: str | None = Field(default=None, max_length=255)
+    is_active: bool | None = Field(default=None)
+
+
+# Database model for Organization
+class Organization(OrganizationBase, table=True):
+    id: int = Field(default=None, primary_key=True)
+
+
+# Properties to return via API
+class OrganizationPublic(OrganizationBase):
+    id: int
+
+
+class OrganizationsPublic(SQLModel):
+    data: list[OrganizationPublic]
+    count: int

backend/app/models/project.py

@@ -0,0 +1,37 @@
+from sqlmodel import Field, Relationship, SQLModel
+
+
+# Shared properties for a Project
+class ProjectBase(SQLModel):
+    name: str = Field(index=True, max_length=255)
+    description: str | None = Field(default=None, max_length=500)
+    is_active: bool = True
+
+
+# Properties to receive via API on creation
+class ProjectCreate(ProjectBase):
+    organization_id: int
+
+
+# Properties to receive via API on update, all are optional
+class ProjectUpdate(SQLModel):
+    name: str | None = Field(default=None, max_length=255)
+    description: str | None = Field(default=None, max_length=500)
+    is_active: bool | None = Field(default=None)
+
+
+# Database model for Project
+class Project(ProjectBase, table=True):
+    id: int = Field(default=None, primary_key=True)
+    organization_id: int = Field(foreign_key="organization.id")
+
+
+# Properties to return via API
+class ProjectPublic(ProjectBase):
+    id: int
+    organization_id: int
+
+
+class ProjectsPublic(SQLModel):
+    data: list[ProjectPublic]
+    count: int

backend/app/tests/api/routes/test_org.py

@@ -0,0 +1,77 @@
+import pytest
+from fastapi.testclient import TestClient
+from sqlmodel import Session, select
+
+from app import crud
+from app.core.config import settings
+from app.core.security import verify_password
+from app.models import User, UserCreate
+from app.tests.utils.utils import random_email, random_lower_string
+from app.models import Organization, OrganizationCreate, OrganizationUpdate
+from app.api.deps import get_db
+from app.main import app
+from app.crud.organization import create_organization, get_organization_by_id
+
+client = TestClient(app)
+
+@pytest.fixture
+def test_organization(db: Session, superuser_token_headers: dict[str, str]):
+    unique_name = f"TestOrg-{random_lower_string()}"
+    org_data = OrganizationCreate(name=unique_name, is_active=True)
+    organization = create_organization(session=db, org_create=org_data)
+    db.commit()
+    return organization
+
+# Test retrieving organizations
+def test_read_organizations(db: Session, superuser_token_headers: dict[str, str]):
+    response = client.get(f"{settings.API_V1_STR}/organizations/", headers=superuser_token_headers)
+    assert response.status_code == 200
+    response_data = response.json()
+    assert "data" in response_data
+    assert isinstance(response_data["data"], list)
+
+# Test creating an organization
+def test_create_organization(db: Session, superuser_token_headers: dict[str, str]):
+    unique_name = f"Org-{random_lower_string()}"
+    org_data = {"name": unique_name, "is_active": True}
+    response = client.post(
+        f"{settings.API_V1_STR}/organizations/", json=org_data, headers=superuser_token_headers
+    )
+
+    assert 200 <= response.status_code < 300
+    created_org = response.json()
+    assert "data" in created_org  # Make sure there's a 'data' field
+    created_org_data = created_org["data"]
+    org = get_organization_by_id(session=db, org_id=created_org_data["id"])
+    assert org is not None  # The organization should be found in the DB
+    assert org.name == created_org_data["name"]
+    assert org.is_active == created_org_data["is_active"]
+
+
+def test_update_organization(db: Session, test_organization: Organization, superuser_token_headers: dict[str, str]):
+    unique_name = f"UpdatedOrg-{random_lower_string()}"  # Ensure a unique name
+    update_data = {"name": unique_name, "is_active": False}
+
+    response = client.patch(
+        f"{settings.API_V1_STR}/organizations/{test_organization.id}",
+        json=update_data,
+        headers=superuser_token_headers,
+    )
+
+    assert response.status_code == 200
+    updated_org = response.json()["data"]    
+    assert "name" in updated_org
+    assert updated_org["name"] == update_data["name"]
+    assert "is_active" in updated_org
+    assert updated_org["is_active"] == update_data["is_active"]
+
+
+# Test deleting an organization
+def test_delete_organization(db: Session, test_organization: Organization, superuser_token_headers: dict[str, str]):
+    response = client.delete(
+        f"{settings.API_V1_STR}/organizations/{test_organization.id}", headers=superuser_token_headers
+    )
+    assert response.status_code == 200
+    response = client.get(f"{settings.API_V1_STR}/organizations/{test_organization.id}", headers=superuser_token_headers)
+    assert response.status_code == 404
+