Warning and Complaint Increment Routes

backend/src/main.py

@@ -4,6 +4,7 @@
 from src.modules.account.account_router import account_router
 from src.modules.location.location_router import location_router
 from src.modules.party.party_router import party_router
+from src.modules.police.police_router import police_router
 from src.modules.student.student_router import student_router
 
 app = FastAPI()
@@ -42,3 +43,4 @@ def read_root():
 app.include_router(party_router)
 app.include_router(student_router)
 app.include_router(location_router)
+app.include_router(police_router)

backend/src/modules/location/location_model.py

@@ -1,8 +1,12 @@
 from typing import Self
 
+from pydantic import AwareDatetime, BaseModel, Field
 from pydantic import AwareDatetime, BaseModel
 from src.core.models import PaginatedResponse
 
+# Maximum allowed value for warning/citation counts to prevent overflow
+MAX_COUNT = 999999
+
 
 class AutocompleteInput(BaseModel):
     # Input for address autocomplete
@@ -32,6 +36,8 @@ class AddressData(BaseModel):
 
 
 class LocationData(AddressData):
+    warning_count: int = Field(default=0, ge=0, le=MAX_COUNT)
+    citation_count: int = Field(default=0, ge=0, le=MAX_COUNT)
     warning_count: int = 0
     citation_count: int = 0
     hold_expiration: AwareDatetime | None = None
@@ -72,6 +78,8 @@ class Location(LocationData):
 
 class LocationCreate(BaseModel):
     google_place_id: str
+    warning_count: int = Field(default=0, ge=0, le=MAX_COUNT)
+    citation_count: int = Field(default=0, ge=0, le=MAX_COUNT)
     warning_count: int = 0
     citation_count: int = 0
     hold_expiration: AwareDatetime | None = None

backend/src/modules/location/location_service.py

@@ -18,7 +18,13 @@
 )
 
 from .location_entity import LocationEntity
-from .location_model import AddressData, AutocompleteResult, Location, LocationData
+from .location_model import (
+    AddressData,
+    AutocompleteResult,
+    Location,
+    LocationData,
+    MAX_COUNT,
+)
 
 
 class GoogleMapsAPIException(InternalServerException):
@@ -64,6 +70,13 @@ def __init__(self, location_id: int, hold_expiration: datetime):
         )
 
 
+class CountLimitExceededException(BadRequestException):
+    def __init__(self, location_id: int, count_type: str):
+        super().__init__(
+            f"Cannot increment {count_type} for location {location_id}: maximum count of {MAX_COUNT} reached"
+        )
+
+
 def get_gmaps_client() -> googlemaps.Client:
     # Dependency injection function for Google Maps client.
     return googlemaps.Client(key=env.GOOGLE_MAPS_API_KEY)
@@ -183,6 +196,28 @@ async def delete_location(self, location_id: int) -> Location:
         await self.session.commit()
         return location
 
+    async def increment_warnings(self, location_id: int) -> Location:
+        """Increment the warning count for a location by 1."""
+        location_entity = await self._get_location_entity_by_id(location_id)
+        if location_entity.warning_count >= MAX_COUNT:
+            raise CountLimitExceededException(location_id, "warning_count")
+        location_entity.warning_count += 1
+        self.session.add(location_entity)
+        await self.session.commit()
+        await self.session.refresh(location_entity)
+        return location_entity.to_model()
+
+    async def increment_citations(self, location_id: int) -> Location:
+        """Increment the citation count for a location by 1."""
+        location_entity = await self._get_location_entity_by_id(location_id)
+        if location_entity.citation_count >= MAX_COUNT:
+            raise CountLimitExceededException(location_id, "citation_count")
+        location_entity.citation_count += 1
+        self.session.add(location_entity)
+        await self.session.commit()
+        await self.session.refresh(location_entity)
+        return location_entity.to_model()
+
     async def autocomplete_address(self, input_text: str) -> list[AutocompleteResult]:
         # Autocomplete an address using Google Maps Places API. Biased towards Chapel Hill, NC area
         try:

backend/src/modules/police/police_router.py

@@ -0,0 +1,44 @@
+from fastapi import APIRouter, Depends, status
+from src.core.authentication import authenticate_police_or_admin
+from src.modules.account.account_model import Account
+from src.modules.location.location_model import Location
+from src.modules.location.location_service import LocationService
+from src.modules.police.police_model import PoliceAccount
+
+police_router = APIRouter(prefix="/api/police", tags=["police"])
+
+
+@police_router.post(
+    "/locations/{location_id}/warnings",
+    response_model=Location,
+    status_code=status.HTTP_200_OK,
+    summary="Increment location warning count",
+    description="Increments the warning count for a location. Requires police or admin authentication.",
+)
+async def increment_warnings(
+    location_id: int,
+    location_service: LocationService = Depends(),
+    _: Account | PoliceAccount = Depends(authenticate_police_or_admin),
+) -> Location:
+    """
+    Increment the warning count for a location by 1.
+    """
+    return await location_service.increment_warnings(location_id)
+
+
+@police_router.post(
+    "/locations/{location_id}/citations",
+    response_model=Location,
+    status_code=status.HTTP_200_OK,
+    summary="Increment location citation count",
+    description="Increments the citation count for a location. Requires police or admin authentication.",
+)
+async def increment_citations(
+    location_id: int,
+    location_service: LocationService = Depends(),
+    _: Account | PoliceAccount = Depends(authenticate_police_or_admin),
+) -> Location:
+    """
+    Increment the citation count for a location by 1.
+    """
+    return await location_service.increment_citations(location_id)