Adding a status db to better track camera status. (#12)

Reviewed-on: #12
Co-authored-by: Andrew Kettel <andrew.kettel@gmail.com>
Co-committed-by: Andrew Kettel <andrew.kettel@gmail.com>

src/database.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from flask_sqlalchemy import SQLAlchemy
+
+db = SQLAlchemy()
+
+
+class CameraStatus(db.Model):
+    __tablename__ = "camera_status"
+
+    id: db.Mapped[int] = db.mapped_column(db.Integer, primary_key=True)
+    running: db.Mapped[bool] = db.mapped_column(
+        db.Boolean, nullable=False, default=False
+    )
+    updated_at: db.Mapped[datetime] = db.mapped_column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )
+
+    @staticmethod
+    def get() -> CameraStatus:
+        """Get the single status row, creating it if it doesn't exist."""
+        status = db.session.get(CameraStatus, 1)
+        if status is None:
+            status = CameraStatus(id=1, running=False)
+            db.session.add(status)
+            db.session.commit()
+        return status
+
+    @staticmethod
+    def set_running(running: bool) -> CameraStatus:
+        status = CameraStatus.get()
+        status.running = running
+        status.updated_at = datetime.now(timezone.utc)
+        db.session.commit()
+        return status
+
+
+class CameraEvent(db.Model):
+    __tablename__ = "camera_events"
+
+    id: db.Mapped[int] = db.mapped_column(db.Integer, primary_key=True)
+    action: db.Mapped[str] = db.mapped_column(
+        db.String(10), nullable=False
+    )  # 'start' | 'stop'
+    ip_address: db.Mapped[str] = db.mapped_column(db.String(45), nullable=False)
+    timestamp: db.Mapped[datetime] = db.mapped_column(
+        db.DateTime(timezone=True),
+        nullable=False,
+        default=lambda: datetime.now(timezone.utc),
+    )
+
+    @staticmethod
+    def log(action: str, ip_address: str) -> CameraEvent:
+        event = CameraEvent(action=action, ip_address=ip_address)
+        db.session.add(event)
+        db.session.commit()
+        return event
+
+    @staticmethod
+    def recent(limit: int = 50) -> list[CameraEvent]:
+        return (
+            db.session.execute(
+                db.select(CameraEvent)
+                .order_by(CameraEvent.timestamp.desc())
+                .limit(limit)
+            )
+            .scalars()
+            .all()
+        )