Add delete sensor/{sensor id}

app/api/dependencies/auth.py

@@ -1,4 +1,5 @@
 import jwt
+from typing import Dict, Union
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 
@@ -11,7 +12,7 @@
 
 
 
-def authenticate_user(username: str, password: str) -> dict[str, str | bool]:
+def authenticate_user(username: str, password: str) -> Dict[str, Union[str, bool]]:
     if settings.ENV == "dev":
         return {"status": "success", "message": "ok", "result": True}
     else:
@@ -47,8 +48,8 @@ async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
 
 
 # Function to decode a JWT token using the specified secret and algorithm
-def unhash(token: str) -> dict[str, str]:
+def unhash(token: str) -> Dict[str, str]:
     return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.ALG]) # type: ignore[no-any-return]
 
-def hash(payload: dict[str, str]) -> str:
+def hash(payload: Dict[str, str]) -> str:
     return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.ALG)

app/api/dependencies/pytas.py

@@ -6,13 +6,14 @@
 from app.pytas.http import TASClient # type: ignore[attr-defined]
 from app.core.config import get_settings
 
+from typing import List
 settings = get_settings()
 
 ENVIRONMENT = settings.ENV
 
 dev_allocations = ["WEATHER-456", "WEATHER-457", "WEATHER-458", "TEST-123", "string"]
 
-def get_allocations(username: str) -> list[str]:
+def get_allocations(username: str) -> List[str]:
     if ENVIRONMENT == "dev":
         return dev_allocations
     else:

app/api/v1/routes/campaigns/campaign_station_sensor_measurements.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Optional
+from typing import List, Optional, Union
 from sqlalchemy.orm import Session
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 
@@ -38,14 +38,14 @@ async def get_sensor_measurements(
     campaign_id: int,
     station_id: int,
     sensor_id: int,
-    start_date: datetime | None = None,
-    end_date: datetime | None = None,
-    min_measurement_value: float | None = None,
-    max_measurement_value: float | None = None,
+    start_date: Optional[datetime] = None,
+    end_date: Optional[datetime] = None,
+    min_measurement_value: Optional[float] = None,
+    max_measurement_value: Optional[float] = None,
     current_user: User = Depends(get_current_user),
     limit: int = 1000,
     page: int = 1,
-    downsample_threshold: int | None = None,
+    downsample_threshold: Optional[int] = None,
     db: Session = Depends(get_db),
 ) -> ListMeasurementsResponsePagination:
     if not check_allocation_permission(current_user, campaign_id):
@@ -54,19 +54,19 @@ async def get_sensor_measurements(
     measurement_service = MeasurementService(measurement_repository)
     return measurement_service.list_measurements(sensor_id=sensor_id, start_date=start_date, end_date=end_date, min_value=min_measurement_value, max_value=max_measurement_value, page=page, limit=limit, downsample_threshold=downsample_threshold)
 
-@router.get("/measurements/confidence-intervals", response_model=list[AggregatedMeasurement])
+@router.get("/measurements/confidence-intervals", response_model=List[AggregatedMeasurement])
 async def get_measurements_with_confidence_intervals(
     campaign_id: int,
     station_id: int,
     sensor_id: int,
     interval: str = Query("hour", description="Time interval for aggregation (minute, hour, day)"),
     interval_value: int = Query(1, description="Multiple of interval (e.g., 15 for 15-minute intervals)"),
-    start_date: datetime | None = Query(None, description="Start date for filtering measurements"),
-    end_date: datetime | None = Query(None, description="End date for filtering measurements"),
-    min_value: float | None = Query(None, description="Minimum measurement value to include"),
-    max_value: float | None = Query(None, description="Maximum measurement value to include"),
+    start_date: Optional[datetime] = Query(None, description="Start date for filtering measurements"),
+    end_date: Optional[datetime] = Query(None, description="End date for filtering measurements"),
+    min_value: Optional[float] = Query(None, description="Minimum measurement value to include"),
+    max_value: Optional[float] = Query(None, description="Maximum measurement value to include"),
     db: Session = Depends(get_db)
-) -> list[AggregatedMeasurement]:
+) -> List[AggregatedMeasurement]:
     """Get sensor measurements with confidence intervals for visualization."""
     measurement_repository = MeasurementRepository(db)
     measurement_service = MeasurementService(measurement_repository)

app/api/v1/routes/campaigns/campaign_station_sensors.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, Union
 from sqlalchemy.orm import Session
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 
@@ -25,10 +25,10 @@ async def list_sensors(
     station_id: int,
     page: int = 1,
     limit: int = 20,
-    variable_name: str | None = Query(None, description="Filter sensors by variable name (partial match)"),
-    units: str | None = Query(None, description="Filter sensors by units (exact match)"),
-    alias: str | None = Query(None, description="Filter sensors by alias (partial match)"),
-    description_contains: str | None = Query(None, description="Filter sensors by text in description (partial match)"),
+    variable_name: Optional[str] = Query(None, description="Filter sensors by variable name (partial match)"),
+    units: Optional[str] = Query(None, description="Filter sensors by units (exact match)"),
+    alias: Optional[str] = Query(None, description="Filter sensors by alias (partial match)"),
+    description_contains: Optional[str] = Query(None, description="Filter sensors by text in description (partial match)"),
     postprocess: Optional[bool] = Query(None, description="Filter sensors by postprocess flag"),
     current_user: User = Depends(get_current_user),
     db: Session = Depends(get_db),
@@ -87,7 +87,7 @@ async def get_sensor(
 
 
 @router.delete("/sensors", status_code=204)
-def delete_sensor(
+def delete_sensors(
     campaign_id: int,
     station_id: int,
     db: Session = Depends(get_db),
@@ -101,6 +101,29 @@ def delete_sensor(
     return Response(status_code=204)
 
 
+@router.delete("/sensors/{sensor_id}", status_code=204)
+def delete_sensor(
+    campaign_id: int,
+    station_id: int,
+    sensor_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+) -> Response:
+    if not check_allocation_permission(current_user, campaign_id):
+        raise HTTPException(status_code=404, detail="Allocation is incorrect")
+
+    sensor_service = SensorService(
+        sensor_repository=SensorRepository(db),
+        measurement_repository=MeasurementRepository(db)
+    )
+
+    success = sensor_service.delete_sensor(sensor_id)
+    if not success:
+        raise HTTPException(status_code=404, detail="Sensor not found")
+
+    return Response(status_code=204)
+
+
 
 @router.put("/sensors/{sensor_id}", response_model=SensorCreateResponse)
 def update_sensor(

app/api/v1/routes/campaigns/campaign_stations.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Annotated
+from typing import Annotated, Optional
 
 from app.services.campaign_service import CampaignService
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
@@ -162,9 +162,9 @@ async def export_measurements_csv(
     campaign_id: int,
     station_id: int,
     start_date: Annotated[
-        datetime | None, Query(description="Start date filter")
+        Optional[datetime], Query(description="Start date filter")
     ] = None,
-    end_date: Annotated[datetime | None, Query(description="End date filter")] = None,
+    end_date: Annotated[Optional[datetime], Query(description="End date filter")] = None,
     current_user: User = Depends(get_current_user),
     db: Session = Depends(get_db),
 ) -> StreamingResponse:

app/api/v1/routes/campaigns/root.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Annotated
+from typing import Annotated, Optional, List
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Response
 from sqlalchemy.orm import Session
@@ -38,19 +38,19 @@ async def list_campaigns(
     page: int = 1,
     limit: int = 20,
     bbox: Annotated[
-        str | None,
+        Optional[str],
         Query(description="Bounding box of the campaign west,south,east,north"),
     ] = None,
     start_date: Annotated[
-        datetime | None,
-        Query(description="Start date of the campaign", example="2024-01-01"),
+        Optional[datetime],
+        Query(description="Start date of the campaign", examples=["2024-01-01"]),
     ] = None,
     end_date: Annotated[
-        datetime | None,
-        Query(description="End date of the campaign", example="2025-01-01"),
+        Optional[datetime],
+        Query(description="End date of the campaign", examples=["2025-01-01"]),
     ] = None,
     sensor_variables: Annotated[
-        list[str] | None, Query(description="List of sensor variables to filter by")
+        Optional[List[str]], Query(description="List of sensor variables to filter by")
     ] = None,
     current_user: User = Depends(get_current_user),
     db: Session = Depends(get_db),
@@ -84,7 +84,7 @@ async def get_campaign(
 
 
 @router.delete("/{campaign_id}", status_code=204)
-def delete_sensor(
+def delete_campaign(
     campaign_id: int,
     db: Session = Depends(get_db),
     current_user: User = Depends(get_current_user),

app/api/v1/routes/projects/projects.py

@@ -4,13 +4,14 @@
 from app.api.v1.schemas.user import User
 from app.services.project_service import ProjectService
 from app.pytas.models.schemas import PyTASProject, PyTASUser
+from typing import List
 router = APIRouter(prefix="", tags=["projects"])
 
 @router.get("/projects")
-async def get_projects(current_user: User = Depends(get_current_user)) -> list[PyTASProject]:
+async def get_projects(current_user: User = Depends(get_current_user)) -> List[PyTASProject]:
     return ProjectService().get_projects_for_user(current_user.username)
 
 
 @router.get("/projects/{project_id}/members")
-async def get_project_members_for_user(project_id: str) -> list[PyTASUser]:
+async def get_project_members_for_user(project_id: str) -> List[PyTASUser]:
     return ProjectService().get_project_members(project_id)

app/api/v1/routes/root.py

@@ -1,3 +1,4 @@
+from typing import Dict
 # mypy: allow-untyped-calls
 
 import jwt

app/api/v1/routes/sensor_variables/sensor_variables.py

@@ -5,11 +5,12 @@
 from app.db.repositories.sensor_repository import SensorRepository
 from app.db.session import get_db
 
+from typing import List
 router = APIRouter(prefix="/sensor_variables", tags=["sensor_variables"])
 
 
 @router.get("")
-async def list_sensor_variables(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)) -> list[str]:
+async def list_sensor_variables(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)) -> List[str]:
     sensor_repository = SensorRepository(db)
     return sensor_repository.list_sensor_variables()
 

app/api/v1/schemas/campaign.py

@@ -8,41 +8,41 @@
 
 class CampaignsIn(BaseModel):
     name: str
-    contact_name: str | None = None
-    contact_email: str | None = None
-    description: str | None = None
-    start_date: datetime | None = None
-    end_date: datetime | None = None
+    contact_name: Optional[str] = None
+    contact_email: Optional[str] = None
+    description: Optional[str] = None
+    start_date: Optional[datetime] = None
+    end_date: Optional[datetime] = None
     allocation: str
 
 class CampaignCreateResponse(BaseModel):
     id: int
 
 class Location(BaseModel):
-    bbox_west: float | None = None
-    bbox_east: float | None = None
-    bbox_south: float | None = None
-    bbox_north: float | None = None
+    bbox_west: Optional[float] = None
+    bbox_east: Optional[float] = None
+    bbox_south: Optional[float] = None
+    bbox_north: Optional[float] = None
 
 class SummaryListCampaigns(BaseModel):
-    sensor_types: List[str] | None = None
-    variable_names: List[str] | None = None
+    sensor_types: Optional[List[str]] = None
+    variable_names: Optional[List[str]] = None
 
 class ListCampaignsResponseItem(BaseModel):
     id: int
     name: str
-    location: Location | None = None
-    description: str | None = None
-    contact_name: str | None = None
-    contact_email: str | None = None
-    start_date: datetime | None = None
-    end_date: datetime | None = None
-    allocation: str | None = None
+    location: Optional[Location] = None
+    description: Optional[str] = None
+    contact_name: Optional[str] = None
+    contact_email: Optional[str] = None
+    start_date: Optional[datetime] = None
+    end_date: Optional[datetime] = None
+    allocation: Optional[str] = None
     summary: SummaryListCampaigns
     geometry: dict = Field(default_factory=dict, nullable=True)  # type: ignore[call-overload,type-arg]
 
 class ListCampaignsResponsePagination(BaseModel):
-    items: list[ListCampaignsResponseItem]
+    items: List[ListCampaignsResponseItem]
     total: int
     page: int
     size: int
@@ -57,13 +57,13 @@ class SummaryGetCampaign(BaseModel):
 class GetCampaignResponse(BaseModel):
     id: int
     name: str
-    description: str | None = None
-    contact_name: str | None = None
-    contact_email: str | None = None
-    start_date: datetime | None = None
-    end_date: datetime | None = None
+    description: Optional[str] = None
+    contact_name: Optional[str] = None
+    contact_email: Optional[str] = None
+    start_date: Optional[datetime] = None
+    end_date: Optional[datetime] = None
     allocation: str
-    location: Location | None = None
+    location: Optional[Location] = None
     summary: SummaryGetCampaign
     geometry: dict = Field(default_factory=dict, nullable=True)  # type: ignore[call-overload,type-arg]
     stations: list[StationsListResponseItem] = []

app/api/v1/schemas/measurement.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Optional, List
+from typing import Optional, List, Union
 
 from geoalchemy2 import Geometry
 from geojson_pydantic import Point
@@ -28,13 +28,13 @@ class MeasurementItem(BaseModel):
     value: float
     geometry: Point
     collectiontime: datetime
-    sensorid: int | None = None
-    variablename: str | None = None  # modified
-    variabletype: str | None = None
-    description: str | None = None
+    sensorid: Optional[int] = None
+    variablename: Optional[str] = None  # modified
+    variabletype: Optional[str] = None
+    description: Optional[str] = None
 
 class ListMeasurementsResponsePagination(BaseModel):
-    items: list[MeasurementItem]
+    items: List[MeasurementItem]
     total: int
     page: int
     size: int
@@ -43,7 +43,7 @@ class ListMeasurementsResponsePagination(BaseModel):
     max_value: float
     average_value: float
     downsampled: bool
-    downsampled_total: int | None = None
+    downsampled_total: Optional[int] = None
 
 class AggregatedMeasurement(BaseModel):
     measurement_time: datetime