God bless our pity souls

Author: Gogobobo11

src/api/api.py

@@ -2,7 +2,7 @@
 from pydantic import BaseModel
 from fastapi.middleware.cors import CORSMiddleware
 
-from .models import CreateCamera, CreateZone
+from .models import *
 
 import cv2
 from fastapi.responses import StreamingResponse
@@ -224,13 +224,7 @@ async def get_camera(camera_id: int):
         async def update_camera(camera_id: int, updated_fields: Request):
             try:
                 updated_fields = await updated_fields.json()
-
-                if not isinstance(updated_fields, dict):
-                    raise HTTPException(
-                        status_code=400,
-                        detail="Request body must be a JSON dict"
-                    )
-            
+                
                 camera = self.db_manager.update_camera(camera_id, updated_fields)
 
                 if camera is None:
@@ -250,17 +244,9 @@ async def update_camera(camera_id: int, updated_fields: Request):
                 )
 
         @self.app.put("/zones/{zone_id}")
-        async def update_zone(zone_id: int, updated_fields: Request):
+        async def update_zone(zone_id: int, update: UpdateZone):
             try:
-                updated_fields = await updated_fields.json()
-                
-                if not isinstance(updated_fields, dict):
-                    raise HTTPException(
-                        status_code=400,
-                        detail="Request body must be a JSON dict"
-                    )
-                
-                zone = self.db_manager.update_zone(zone_id, updated_fields)
+                zone = self.db_manager.update_zone(zone_id, update)
 
                 if zone is None:
                     raise HTTPException(

src/api/models.py

@@ -1,5 +1,5 @@
-from typing import List, Dict, TypedDict, Any
-from pydantic import BaseModel, field_validator
+from typing import List, Dict, TypedDict, Any, Optional, Literal
+from pydantic import BaseModel, field_validator, Field, ValidationError
 import json
 
 class CreateCamera(BaseModel):
@@ -15,7 +15,7 @@ class CreateCamera(BaseModel):
     @classmethod
     def validate_title(cls, title):
         if len(title) < 1 or len(title) > 200:
-            raise ValueError(f"Invalid camera title: {title}")
+            raise ValidationError(f"Invalid camera title: {title}")
         return title
 
     @field_validator('source')
@@ -27,28 +27,28 @@ def validate_source(cls, source):
     @classmethod
     def validate_latitude(cls, latitude):
         if latitude > 90 or latitude < -90:
-            raise ValueError(f"Invalid latitude value: {latitude}")
+            raise ValidationError(f"Invalid latitude value: {latitude}")
         return latitude
 
     @field_validator('longitude')
     @classmethod
     def validate_longitude(cls, longitude):
         if longitude > 180 or longitude < -180:
-            raise ValueError(f"Invalid longitude value: {longitude}")
+            raise ValidationError(f"Invalid longitude value: {longitude}")
         return longitude
 
     @field_validator('image_width')
     @classmethod
     def validate_image_width(cls, image_width):
         if image_width <= 0: 
-            raise ValueError(f"Invalid image_width value: {image_width}")
+            raise ValidationError(f"Invalid image_width value: {image_width}")
         return image_width
 
     @field_validator('image_height')
     @classmethod
     def validate_image_height(cls, image_height):
         if image_height <= 0: 
-            raise ValueError(f"Invalid image_height value: {image_height}")
+            raise ValidationError(f"Invalid image_height value: {image_height}")
         return image_height
 
     @field_validator('calib')
@@ -57,99 +57,52 @@ def validate_calib(cls, calib):
         if calib is not None:
             try:
                 json.dumps(calib)
-            except (TypeError, ValueError) as e:
-                raise ValueError(f"Invalid calibration data: {e}")
+            except:
+                raise ValidationError(f"Invalid calibration data")
         return calib
 
 class Point(BaseModel):
-    latitude: float
-    longitude: float
-    x: int
-    y: int
+    latitude: float = Field(ge=-90, le=90)
+    longitude: float = Field(ge=-180, le=180)
+    x: int = Field(ge=0)
+    y: int = Field(ge=0)
 
     def __eq__(self, other):
         if not isinstance(other, Point):
             return False
         return (self.x == other.x and self.y == other.y)
 
-    @field_validator('latitude')
-    @classmethod
-    def validate_latitude(cls, latitude):
-        if latitude > 90 or latitude < -90:
-            raise ValueError(f"Invalid latitude value: {latitude}")
-        return latitude
-    
-    @field_validator('longitude')
-    @classmethod
-    def validate_longitude(cls, longitude):
-        if longitude > 180 or longitude < -180:
-            raise ValueError(f"Invalid longitude value: {longitude}")
-        return longitude
-    
-    @field_validator('x')
-    @classmethod
-    def validate_x(cls, x):
-        if x < 0: 
-            raise ValueError(f"Invalid x value: {x}")
-        
-        return x
+class ZoneBase(BaseModel):
+    camera_id: Optional[int] = Field(None, ge=1)
+    zone_type: Optional[Literal['parallel', 'standard']] = None
+    capacity: Optional[int] = Field(None, gt=0)
+    pay: Optional[int] = Field(None, ge=0)
+    points: Optional[List[Point]] = None
 
-    @field_validator('y')
+    @field_validator('points')
     @classmethod
-    def validate_y(cls, y):
-        if y < 0: 
-            raise ValueError(f"Invalid y value: {y}")
+    def validate_points(cls, v: Optional[List[Point]]) -> Optional[List[Point]]:
+        if v is None:
+            return v
+            
+        if len(v) != 4:
+            raise ValidationError(f"Invalid points count: {len(v)}. Must be exactly 4 points")
 
-        return y
+        for lhs in range(0, 4):
+            for rhs in range(lhs + 1, 4):
+                if v[lhs] == v[rhs]:
+                    raise ValidationError(f"Degenerate rectangle")
+            
+        return v
 
-class CreateZone(BaseModel):
+# 3. Модели запросов через наследование
+class CreateZone(ZoneBase):
     camera_id: int
-    zone_type: str
-    capacity: int
-    pay: int
+    zone_type: Literal['parallel', 'standard']
+    capacity: int = Field(gt=0)
+    pay: int = Field(ge=0)
     points: List[Point]
 
-    @field_validator('camera_id')
-    @classmethod
-    def validate_camera_id(cls, camera_id):
-        if camera_id <= 0:
-            raise ValueError(f"Invalid camera_id value: {camera_id}")
-        
-        return camera_id
-    
-    @field_validator('zone_type')
-    @classmethod
-    def validate_zone_type(cls, zone_type):
-        if zone_type not in ['parallel', 'standard']:
-            raise ValueError(f"Invalid zone_type value: {zone_type}")
-        
-        return zone_type
-    
-    @field_validator('capacity')
-    @classmethod
-    def validate_capacity(cls, capacity):
-        if capacity <= 0:
-            raise ValueError(f"Invalid capacity value: {capacity}")
-        
-        return capacity
-        
-    @field_validator('pay')
-    @classmethod
-    def validate_pay(cls, pay):
-        if pay < 0:
-            raise ValueError(f"Invalid pay value: {pay}")
-        
-        return pay
-    
-    @field_validator('points')
-    @classmethod
-    def validate_points(cls, points):
-        if len(points) != 4:
-            raise ValueError(f"Invalid points count: {len(points)}")
-        
-        for lhs in range(0, len(points)):
-            for rhs in range(lhs + 1, len(points)):
-                if points[lhs] == points[rhs]:
-                    raise ValueError(f"Degenerate rectangle")
-        
-        return points
+class UpdateZone(ZoneBase):
+    occupied: Optional[int] = Field(None, ge=0)
+    confidence: Optional[float] = Field(None, ge=0, le=1)

src/db_manager/db_manager.py

@@ -1,9 +1,10 @@
 import os
-from sqlalchemy import create_engine, inspect, text, func, update, select
+from sqlalchemy import create_engine, inspect, text, func, update, select, delete
 from sqlalchemy.orm import sessionmaker, Session, joinedload
 from sqlalchemy.exc import SQLAlchemyError
 import contextlib
 from typing import Generator, Optional
+from fastapi import HTTPException
 
 from .models import Base, Camera, ParkingZone, ParkingZonePoint, datetime, timezone
 
@@ -120,6 +121,31 @@ def _create_tables(self):
             print(f"Error creating tables: {e}")
             raise
 
+    def _update_points(self, zone_id, points):
+        if not isinstance(points, list[dict]) or len(points) != 4:
+            raise HTTPException(
+                status_code=400,
+                detail="Invalid points request format"
+            )
+        
+        with self.get_session() as session:
+            stmt = delete(ParkingZonePoint).where(ParkingZonePoint.parking_zone_id == zone_id)
+            session.execute(stmt)
+            
+            for point in points:
+                stmt = update(ParkingZonePoint).where(ParkingZonePoint.id == camera_id)
+
+                stmt = stmt.values(updated_fields)
+
+                session.execute(stmt)
+
+                session.commit()
+
+                camera = session.query(Camera).filter(Camera.id == camera_id).one_or_none()
+
+            return camera.serialize() if camera is not None else None
+
+
     @contextlib.contextmanager
     def get_session(self) -> Generator[Session, None, None]:
         """Контекстный менеджер для получения сессии"""
@@ -277,6 +303,12 @@ def get_camera(self, camera_id):
             return camera.serialize() if camera is not None else None
 
     def update_camera(self, camera_id, updated_fields):
+        if not isinstance(updated_fields, dict):
+            raise HTTPException(
+                status_code=400,
+                detail="Request body must be a JSON dict"
+            )
+        
         with self.get_session() as session:
             stmt = update(Camera).where(Camera.id == camera_id)
 
@@ -290,16 +322,38 @@ def update_camera(self, camera_id, updated_fields):
 
             return camera.serialize() if camera is not None else None
 
-    def update_zone(self, zone_id, updated_fields):
+    def update_zone(self, zone_id, update):
+        if self.get_zone(zone_id) is None: return None
+
         with self.get_session() as session:
+
+            if update.points is not None:
+                session.execute(
+                    delete(ParkingZonePoint).where(ParkingZonePoint.parking_zone_id == zone_id)
+                )
+
+                for point in update.points: 
+                    session.add(
+                        ParkingZonePoint(
+                            parking_zone_id=zone_id,
+                            latitude=point.latitude,
+                            longitude=point.longitude,
+                            x=point.x,
+                            y=point.y
+                        )
+                    )
+
             stmt = update(ParkingZone).where(ParkingZone.id == zone_id)
-            if "capacity" in updated_fields.keys():
-                updated_fields['parking_lots_count'] = updated_fields.pop('capacity')
+
+            update_dump = update.model_dump(
+                exclude_none=True,
+                exclude={'points'}
+            )
 
             stmt = stmt.values(
-                updated_fields | {"updated_at": datetime.now(timezone.utc)}
-                    if "occupied" not in updated_fields else 
-                updated_fields | {"occupancy_updated_at": datetime.now(timezone.utc)})
+                update_dump | {"updated_at": datetime.now(timezone.utc)}
+                    if "occupied" not in update_dump else 
+                update_dump | {"occupancy_updated_at": datetime.now(timezone.utc)})
 
             session.execute(stmt)