WIP compliance

backend/app/database.py

@@ -2,7 +2,7 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, declarative_base
 import os
 
-DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./drugs.db")
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./data/drugs.db")
 
 # For SQLite, ensure the directory exists
 if "sqlite" in DATABASE_URL:
@@ -18,10 +18,6 @@ engine = create_engine(
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 Base = declarative_base()
 
-# Drop and recreate all tables (for development only)
-Base.metadata.drop_all(bind=engine)
-Base.metadata.create_all(bind=engine)
-
 def get_db():
     db = SessionLocal()
     try:

backend/app/migrate_compliance.py

@@ -0,0 +1,83 @@
+"""
+Compliance schema migration helpers.
+
+This module applies additive migrations for SQLite databases used by this project.
+It is intentionally lightweight and idempotent because the project does not yet
+use Alembic-style versioned migrations.
+"""
+
+import os
+import sqlite3
+from pathlib import Path
+
+
+DEFAULT_DB_URL = "sqlite:///./data/drugs.db"
+
+
+def _resolve_sqlite_path(db_url: str) -> Path | None:
+    if not db_url.startswith("sqlite:///"):
+        print(f"Unsupported database URL for compliance migration: {db_url}")
+        return None
+
+    raw_path = db_url.replace("sqlite:///", "")
+    if raw_path.startswith("/"):
+        return Path(raw_path)
+    return Path(raw_path)
+
+
+def _column_exists(cursor: sqlite3.Cursor, table_name: str, column_name: str) -> bool:
+    cursor.execute(f"PRAGMA table_info({table_name})")
+    columns = [row[1] for row in cursor.fetchall()]
+    return column_name in columns
+
+
+def _table_exists(cursor: sqlite3.Cursor, table_name: str) -> bool:
+    cursor.execute(
+        "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
+        (table_name,),
+    )
+    return cursor.fetchone() is not None
+
+
+def migrate_compliance_schema() -> None:
+    """Apply additive schema changes needed for compliance features."""
+    db_url = os.getenv("DATABASE_URL", DEFAULT_DB_URL)
+    db_path = _resolve_sqlite_path(db_url)
+    if db_path is None:
+        return
+
+    if not db_path.exists():
+        print(f"Database does not exist at {db_path}, skipping compliance migration")
+        return
+
+    print(f"Running compliance migration on {db_path}")
+
+    conn = sqlite3.connect(str(db_path))
+    cursor = conn.cursor()
+
+    try:
+        if _table_exists(cursor, "drugs") and not _column_exists(cursor, "drugs", "is_controlled"):
+            cursor.execute("ALTER TABLE drugs ADD COLUMN is_controlled BOOLEAN NOT NULL DEFAULT 0")
+            print("Added drugs.is_controlled")
+
+        if _table_exists(cursor, "dispensings") and not _column_exists(cursor, "dispensings", "batch_id"):
+            cursor.execute("ALTER TABLE dispensings ADD COLUMN batch_id INTEGER")
+            print("Added dispensings.batch_id")
+
+        if _table_exists(cursor, "dispensings") and not _column_exists(cursor, "dispensings", "actor_user_id"):
+            cursor.execute("ALTER TABLE dispensings ADD COLUMN actor_user_id INTEGER")
+            print("Added dispensings.actor_user_id")
+
+        # Seed default locations once table exists (created via SQLAlchemy create_all).
+        if _table_exists(cursor, "locations"):
+            cursor.execute("INSERT OR IGNORE INTO locations(name, is_active) VALUES ('Cupboard', 1)")
+            cursor.execute("INSERT OR IGNORE INTO locations(name, is_active) VALUES ('Fridge', 1)")
+            print("Ensured default locations exist")
+
+        conn.commit()
+        print("Compliance migration completed")
+    except sqlite3.Error as exc:
+        conn.rollback()
+        raise RuntimeError(f"Compliance migration failed: {exc}") from exc
+    finally:
+        conn.close()

backend/app/models.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Boolean
+from sqlalchemy import Column, Integer, String, Float, DateTime, Date, ForeignKey, Boolean, Text
 from sqlalchemy.sql import func
 from .database import Base
 
@@ -18,6 +18,7 @@ class Drug(Base):
     id = Column(Integer, primary_key=True, index=True)
     name = Column(String, unique=True, index=True, nullable=False)
     description = Column(String, nullable=True)
+    is_controlled = Column(Boolean, nullable=False, default=False)
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     updated_at = Column(DateTime(timezone=True), onupdate=func.now(), server_default=func.now())
 
@@ -40,8 +41,57 @@ class Dispensing(Base):
 
     id = Column(Integer, primary_key=True, index=True)
     drug_variant_id = Column(Integer, ForeignKey("drug_variants.id"), nullable=False)
+    batch_id = Column(Integer, ForeignKey("batches.id"), nullable=True)
+    actor_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
     quantity = Column(Float, nullable=False)
     animal_name = Column(String, nullable=True)  # Name/ID of the animal (optional)
     user_name = Column(String, nullable=False)  # User who dispensed
     dispensed_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)
     notes = Column(String, nullable=True)
+
+
+class Location(Base):
+    __tablename__ = "locations"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, unique=True, index=True, nullable=False)
+    is_active = Column(Boolean, nullable=False, default=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now(), server_default=func.now())
+
+
+class Batch(Base):
+    __tablename__ = "batches"
+
+    id = Column(Integer, primary_key=True, index=True)
+    drug_variant_id = Column(Integer, ForeignKey("drug_variants.id"), nullable=False, index=True)
+    batch_number = Column(String, nullable=False, index=True)
+    quantity = Column(Float, nullable=False, default=0)
+    expiry_date = Column(Date, nullable=False, index=True)
+    location_id = Column(Integer, ForeignKey("locations.id"), nullable=False, index=True)
+    received_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now(), index=True)
+    notes = Column(String, nullable=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), onupdate=func.now(), server_default=func.now())
+
+
+class DispensingAllocation(Base):
+    __tablename__ = "dispensing_allocations"
+
+    id = Column(Integer, primary_key=True, index=True)
+    dispensing_id = Column(Integer, ForeignKey("dispensings.id"), nullable=False, index=True)
+    batch_id = Column(Integer, ForeignKey("batches.id"), nullable=False, index=True)
+    quantity = Column(Float, nullable=False)
+
+
+class AuditLog(Base):
+    __tablename__ = "audit_logs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    action = Column(String, nullable=False, index=True)
+    entity_type = Column(String, nullable=False, index=True)
+    entity_id = Column(Integer, nullable=True, index=True)
+    actor_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
+    actor_username = Column(String, nullable=False, index=True)
+    details = Column(Text, nullable=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)