Change: DBManager now works on sqlalchemy instead of sqlite3

Author: Gogobobo11

src/api/api.py

@@ -9,12 +9,11 @@ class URL(BaseModel):
 
 class PublicAPI:
     title = "API Server"
-    version = "0.0.1"
+    version = "0.1.0"
     description="ParkTrack server API built with FastAPI"
 
     def __init__(self, db_manager):
         self.db_manager = db_manager
-        self.db_manager.connect()
 
         self.app = FastAPI(
             title=self.title,

src/db_manager/db_manager.py

@@ -1,14 +1,119 @@
-import sqlite3
+from sqlalchemy import create_engine, inspect
+from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.exc import SQLAlchemyError
+import contextlib
+from typing import Generator, Optional
+import os
+
+from .models import Base
 
 class DBManager:
-    def __init__(self, db_name):
-        self.db_name = db_name
-        self.conn = None
-        self.cursor = None
-    
-    def connect(self):
-        self.conn = sqlite3.connect(self.db_name)
-        self.cursor = self.conn.cursor()
+    def __init__(self, database_url: Optional[str] = None):
+        self.database_url = database_url or self._get_default_database_url()
+        self.engine = None
+        self.SessionLocal = None
+        self._initialize_database()
+
+    def _get_default_database_url(self) -> str:
+        """Захардкодил URL базы данных по умолчанию"""
+        # Для SQLite
+        return "sqlite:///./parktrack.db"
+        
+        # Для PostgreSQL:
+        # return "postgresql://user:password@localhost/parktrack"
+
+    def _initialize_database(self):
+        """Инициализация движка и сессии"""
+        try:
+            # Для SQLite нужно special pragma для Foreign Keys
+            connect_args = {}
+            if self.database_url.startswith("sqlite"):
+                connect_args = {"check_same_thread": False}
+            
+            self.engine = create_engine(
+                self.database_url,
+                connect_args=connect_args,
+                echo=True,  # Для тестирования
+                pool_pre_ping=True # Загадочно
+            )
+            
+            self.SessionLocal = sessionmaker(
+                autocommit=False,
+                autoflush=False,
+                bind=self.engine
+            )
+            
+            if not self._check_tables_exist():
+                print(f"Gotta setup database real quick hold on...")
+                self._create_tables()
+            
+            print(f"Database initialized successfully: {self.database_url}")
+            
+        except Exception as e:
+            print(f"Database initialization failed: {e}")
+            raise
+
+    def _check_tables_exist(self) -> bool:
+        try:
+            inspector = inspect(self.engine)
+            existing_tables = inspector.get_table_names()
+            
+            # Получаем список таблиц, которые должны быть созданы
+            required_tables = Base.metadata.tables.keys()
+            
+            print(f"Existing tables: {existing_tables}")
+            print(f"Required tables: {list(required_tables)}")
+            
+            # Проверяем, что ВСЕ нужные таблицы существуют
+            for table_name in required_tables:
+                if table_name not in existing_tables:
+                    print(f"Table {table_name} not found")
+                    return False
+                
+            return True
+            
+        except Exception as e:
+            print(f"Error checking tables: {e}")
+            return False
+
+    def _create_tables(self):
+        try:
+            print("Creating database tables...")
+            Base.metadata.create_all(bind=self.engine)
+            print("Tables created successfully")
+        except Exception as e:
+            print(f"Error creating tables: {e}")
+            raise
+
+    @contextlib.contextmanager
+    def get_session(self) -> Generator[Session, None, None]:
+        """Контекстный менеджер для получения сессии"""
+        session = self.SessionLocal()
+        try:
+            yield session
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
 
     def check_connection(self) -> bool:
-        return True
+        """Проверить соединение с базой данных"""
+        try:
+            with self.get_session() as session:
+                # Простой запрос для проверки соединения
+                session.execute("SELECT 1")
+            return True
+        except SQLAlchemyError as e:
+            print(f"Database connection check failed: {e}")
+            return False
+
+    def get_db(self) -> Generator[Session, None, None]:
+        return self.get_session()
+
+    def execute_raw_sql(self, query: str, params: dict = None):
+        """Выполнить сырой SQL запрос"""
+        with self.get_session() as session:
+            result = session.execute(query, params or {})
+            return result

src/db_manager/models.py

@@ -0,0 +1,78 @@
+from sqlalchemy import Column, Integer, String, Boolean, DateTime, Numeric, ForeignKey
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship
+from datetime import datetime, timezone
+
+Base = declarative_base()
+
+class Camera(Base):
+    __tablename__ = 'cameras'
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    title = Column(String(120))
+    is_active = Column(Boolean, default=False)
+    pixels_height = Column(Integer)
+    pixels_width = Column(Integer)
+    created_at = Column(DateTime, default=timezone.utc)
+    updated_at = Column(DateTime, default=datetime.now(timezone.utc), onupdate=timezone.utc)
+    
+    parking_zones = relationship("ParkingZone", back_populates="camera")
+    cars = relationship("Car", back_populates="camera")
+    
+    def __repr__(self):
+        return f"<Camera(id={self.id}, title='{self.title}', is_active={self.is_active})>"
+
+class ParkingZone(Base):
+    __tablename__ = 'parking_zones'
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    zone_type = Column(String(50))
+    parking_lots_count = Column(Integer)
+    camera_id = Column(Integer, ForeignKey('cameras.id'))
+    
+    camera = relationship("Camera", back_populates="parking_zones")
+    points = relationship("ParkingZonePoint", back_populates="parking_zone")
+    
+    def __repr__(self):
+        return f"<ParkingZone(id={self.id}, zone_type='{self.zone_type}', camera_id={self.camera_id})>"
+
+class ParkingZonePoint(Base):
+    __tablename__ = 'parking_zones_points'
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    parking_zone_id = Column(Integer, ForeignKey('parking_zones.id'))
+    x_component = Column(Numeric(6, 5))
+    y_component = Column(Numeric(6, 5))
+    latitude = Column(Numeric(11, 8))
+    longitude = Column(Numeric(11, 8))
+    
+    parking_zone = relationship("ParkingZone", back_populates="points")
+    
+    def __repr__(self):
+        return f"<ParkingZonePoint(id={self.id}, zone_id={self.parking_zone_id}, x={self.x_component}, y={self.y_component})>"
+
+class Car(Base):
+    __tablename__ = 'cars'
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    confidence_rate = Column(Numeric(5, 4))
+    camera_id = Column(Integer, ForeignKey('cameras.id'))
+    
+    camera = relationship("Camera", back_populates="cars")
+    points = relationship("CarPoint", back_populates="car")
+    
+    def __repr__(self):
+        return f"<Car(id={self.id}, confidence={self.confidence_rate}, camera_id={self.camera_id})>"
+
+class CarPoint(Base):
+    __tablename__ = 'car_points'
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    car_id = Column(Integer, ForeignKey('cars.id'))
+    x_component = Column(Numeric(6, 5))
+    y_component = Column(Numeric(6, 5))
+    
+    car = relationship("Car", back_populates="points")
+    
+    def __repr__(self):
+        return f"<CarPoint(id={self.id}, car_id={self.car_id}, x={self.x_component}, y={self.y_component})>"

src/main.py

@@ -2,7 +2,7 @@
 from db_manager.db_manager import DBManager
 
 def main():
-    api_server = PublicAPI(db_manager=DBManager("mock"))
+    api_server = PublicAPI(DBManager())
     api_server.run(URL(host="127.0.0.1", port=8000))
 
 if __name__ == "__main__":