jamesp/mt-drugs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Barcode scanning and GTIN mapping

Browse Source
This commit is contained in:
James Pattinson
2026-04-16 15:32:36 -04:00
parent 2aeba2f563
commit cfb08bd288
6 changed files with 938 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/main.py
+119 -24
View File
@@ -18,11 +18,13 @@ from .models import (
Batch,
AuditLog,
User,
GtinMapping,
)
from .auth import hash_password, verify_password, create_access_token, get_current_user, get_current_admin_user, get_current_non_readonly_user, ACCESS_TOKEN_EXPIRE_MINUTES
from .mqtt_service import publish_label_print_with_response
from .migrate_to_roles import migrate_users_table
from .migrate_compliance import migrate_compliance_schema
from .migrate_gtin import migrate_gtin_schema
from pydantic import BaseModel
# Run migration to convert is_admin to role
@@ -39,6 +41,11 @@ except Exception as e:
# Create tables
Base.metadata.create_all(bind=engine)
try:
migrate_gtin_schema()
except Exception as e:
print(f"Warning: GTIN migration failed: {e}. Continuing anyway...")
# Seed default locations after table creation.
try:
migrate_compliance_schema()
@@ -138,14 +145,37 @@ class BatchDisposeRequest(BaseModel):
notes: Optional[str] = None
class GtinMappingCreate(BaseModel):
gtin: str
drug_variant_id: int
variant_pack_id: int
class GtinMappingResponse(BaseModel):
id: int
gtin: str
drug_variant_id: int
variant_pack_id: int
drug_id: int
drug_name: str
variant_strength: str
variant_unit: str
pack_label: str
pack_size_in_base_units: float
pack_unit_name: str
class Config:
from_attributes = True
class BatchResponse(BaseModel):
id: int
drug_variant_id: int
batch_number: str
quantity: float
received_pack_id: Optional[int] = None
received_pack_label: Optional[str] = None
received_pack_count: Optional[float] = None
received_pack_unit_name: Optional[str] = None
received_pack_size_snapshot: Optional[float] = None
current_full_pack_count: Optional[float] = None
current_loose_base_units: Optional[float] = None
@@ -187,14 +217,12 @@ class DrugVariantUpdate(BaseModel):
class VariantPackCreate(BaseModel):
label: str
pack_unit_name: str
pack_size_in_base_units: float
is_active: bool = True
class VariantPackUpdate(BaseModel):
label: Optional[str] = None
pack_unit_name: Optional[str] = None
pack_size_in_base_units: Optional[float] = None
is_active: Optional[bool] = None
@@ -203,7 +231,6 @@ class VariantPackUpdate(BaseModel):
class VariantPackResponse(BaseModel):
id: int
drug_variant_id: int
label: str
pack_unit_name: str
pack_size_in_base_units: float
is_active: bool
@@ -383,7 +410,7 @@ def serialize_batch_response(db: Session, batch: Batch) -> Dict[str, Any]:
"batch_number": batch.batch_number,
"quantity": batch.quantity,
"received_pack_id": batch.received_pack_id,
"received_pack_label": pack.label if pack else None,
"received_pack_unit_name": pack.pack_unit_name if pack else None,
"received_pack_count": batch.received_pack_count,
"received_pack_size_snapshot": batch.received_pack_size_snapshot,
"current_full_pack_count": batch.current_full_pack_count,
@@ -429,7 +456,6 @@ def serialize_variant_pack(pack: VariantPack) -> Dict[str, Any]:
return {
"id": pack.id,
"drug_variant_id": pack.drug_variant_id,
"label": pack.label,
"pack_unit_name": pack.pack_unit_name,
"pack_size_in_base_units": pack.pack_size_in_base_units,
"is_active": pack.is_active,
@@ -1050,15 +1076,12 @@ def create_drug_variant(drug_id: int, variant: DrugVariantCreate, db: Session =
db.flush()
# Ensure each variant has at least one active default 1:1 pack representation.
db.add(
VariantPack(
db.add(VariantPack(
drug_variant_id=db_variant.id,
label=f"1 {base_unit}",
pack_unit_name=base_unit,
pack_size_in_base_units=1,
is_active=True,
)
)
))
write_audit_log(
db,
@@ -1232,10 +1255,7 @@ def create_variant_pack(
if not variant:
raise HTTPException(status_code=404, detail="Drug variant not found")
label = payload.label.strip()
pack_unit_name = payload.pack_unit_name.strip()
if not label:
raise HTTPException(status_code=400, detail="Pack label cannot be empty")
if not pack_unit_name:
raise HTTPException(status_code=400, detail="Pack unit name cannot be empty")
if payload.pack_size_in_base_units <= 0:
@@ -1243,7 +1263,6 @@ def create_variant_pack(
row = VariantPack(
drug_variant_id=variant_id,
label=label,
pack_unit_name=pack_unit_name,
pack_size_in_base_units=payload.pack_size_in_base_units,
is_active=payload.is_active,
@@ -1257,7 +1276,6 @@ def create_variant_pack(
actor=current_user,
details={
"variant_id": variant_id,
"label": label,
"pack_unit_name": pack_unit_name,
"pack_size_in_base_units": payload.pack_size_in_base_units,
"is_active": payload.is_active,
@@ -1281,12 +1299,6 @@ def update_variant_pack(
before = serialize_variant_pack(row)
if payload.label is not None:
cleaned = payload.label.strip()
if not cleaned:
raise HTTPException(status_code=400, detail="Pack label cannot be empty")
row.label = cleaned
if payload.pack_unit_name is not None:
cleaned = payload.pack_unit_name.strip()
if not cleaned:
@@ -2200,7 +2212,7 @@ def report_batch_attention(
"location": location.name,
"expiry_date": batch.expiry_date,
"status": "expired",
"received_pack_label": None,
"received_pack_unit_name": None,
"current_full_pack_count": batch.current_full_pack_count,
"current_loose_base_units": batch.current_loose_base_units,
"is_controlled": bool(drug.is_controlled),
@@ -2450,5 +2462,88 @@ def print_notes(notes_request: NotesPrintRequest, current_user: User = Depends(g
detail=f"Error sending notes print request: {str(e)}"
)
@router.get("/gtin/{gtin}", response_model=GtinMappingResponse)
def get_gtin_mapping(
gtin: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
mapping = db.query(GtinMapping).filter(GtinMapping.gtin == gtin).first()
if not mapping:
raise HTTPException(status_code=404, detail="GTIN not found")
variant = db.query(DrugVariant).filter(DrugVariant.id == mapping.drug_variant_id).first()
drug = db.query(Drug).filter(Drug.id == variant.drug_id).first() if variant else None
pack = db.query(VariantPack).filter(VariantPack.id == mapping.variant_pack_id).first()
if not variant or not drug or not pack:
# Referenced records no longer exist — delete the stale mapping and treat as unknown
db.delete(mapping)
db.commit()
raise HTTPException(status_code=404, detail="GTIN not found")
return GtinMappingResponse(
id=mapping.id,
gtin=mapping.gtin,
drug_variant_id=mapping.drug_variant_id,
variant_pack_id=mapping.variant_pack_id,
drug_id=drug.id,
drug_name=drug.name,
variant_strength=variant.strength,
variant_unit=variant.unit,
pack_label=f"{pack.pack_unit_name} of {int(pack.pack_size_in_base_units) if pack.pack_size_in_base_units == int(pack.pack_size_in_base_units) else pack.pack_size_in_base_units}",
pack_size_in_base_units=pack.pack_size_in_base_units,
pack_unit_name=pack.pack_unit_name,
)
@router.post("/gtin", response_model=GtinMappingResponse, status_code=201)
def create_gtin_mapping(
body: GtinMappingCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_non_readonly_user),
):
existing = db.query(GtinMapping).filter(GtinMapping.gtin == body.gtin).first()
if existing:
raise HTTPException(status_code=409, detail="GTIN mapping already exists")
variant = db.query(DrugVariant).filter(DrugVariant.id == body.drug_variant_id).first()
if not variant:
raise HTTPException(status_code=404, detail="Drug variant not found")
pack = db.query(VariantPack).filter(
VariantPack.id == body.variant_pack_id,
VariantPack.drug_variant_id == body.drug_variant_id,
).first()
if not pack:
raise HTTPException(status_code=404, detail="Pack not found for this variant")
drug = db.query(Drug).filter(Drug.id == variant.drug_id).first()
mapping = GtinMapping(
gtin=body.gtin,
drug_variant_id=body.drug_variant_id,
variant_pack_id=body.variant_pack_id,
created_by_user_id=current_user.id,
)
db.add(mapping)
db.commit()
db.refresh(mapping)
return GtinMappingResponse(
id=mapping.id,
gtin=mapping.gtin,
drug_variant_id=mapping.drug_variant_id,
variant_pack_id=mapping.variant_pack_id,
drug_id=drug.id,
drug_name=drug.name,
variant_strength=variant.strength,
variant_unit=variant.unit,
pack_label=f"{pack.pack_unit_name} of {int(pack.pack_size_in_base_units) if pack.pack_size_in_base_units == int(pack.pack_size_in_base_units) else pack.pack_size_in_base_units}",
pack_size_in_base_units=pack.pack_size_in_base_units,
pack_unit_name=pack.pack_unit_name,
)
# Include router with /api prefix
app.include_router(router)
backend/app/migrate_gtin.py
+103
View File
@@ -0,0 +1,103 @@
"""
GTIN mapping table migration.
Creates the gtin_mappings table if it does not already exist.
Idempotent and safe to run on every startup.
"""
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 GTIN migration: {db_url}")
return None
raw_path = db_url.replace("sqlite:///", "")
if raw_path.startswith("/"):
return Path(raw_path)
return Path(raw_path)
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 _column_exists(cursor: sqlite3.Cursor, table_name: str, column_name: str) -> bool:
cursor.execute(f"PRAGMA table_info({table_name})")
return any(row[1] == column_name for row in cursor.fetchall())
def migrate_gtin_schema() -> None:
"""Create gtin_mappings table if it does not exist, and drop label from variant_packs."""
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 GTIN migration")
return
print(f"Running GTIN migration on {db_path}")
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
try:
if not _table_exists(cursor, "gtin_mappings"):
cursor.execute("""
CREATE TABLE gtin_mappings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
gtin VARCHAR(14) NOT NULL,
drug_variant_id INTEGER NOT NULL REFERENCES drug_variants(id),
variant_pack_id INTEGER NOT NULL REFERENCES variant_packs(id),
created_by_user_id INTEGER REFERENCES users(id),
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("CREATE UNIQUE INDEX ix_gtin_mappings_gtin ON gtin_mappings (gtin)")
cursor.execute("CREATE INDEX ix_gtin_mappings_drug_variant_id ON gtin_mappings (drug_variant_id)")
cursor.execute("CREATE INDEX ix_gtin_mappings_variant_pack_id ON gtin_mappings (variant_pack_id)")
print("Created gtin_mappings table")
else:
print("gtin_mappings table already exists, skipping")
# Drop label column from variant_packs if it still exists
if _table_exists(cursor, "variant_packs") and _column_exists(cursor, "variant_packs", "label"):
print("Dropping label column from variant_packs")
cursor.execute("""
CREATE TABLE variant_packs_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
drug_variant_id INTEGER NOT NULL REFERENCES drug_variants(id),
pack_unit_name VARCHAR NOT NULL DEFAULT 'pack',
pack_size_in_base_units FLOAT NOT NULL DEFAULT 1,
is_active BOOLEAN NOT NULL DEFAULT 1,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
cursor.execute("""
INSERT INTO variant_packs_new (id, drug_variant_id, pack_unit_name, pack_size_in_base_units, is_active, created_at, updated_at)
SELECT id, drug_variant_id, pack_unit_name, pack_size_in_base_units, is_active, created_at, updated_at
FROM variant_packs
""")
# Re-create indexes
cursor.execute("DROP INDEX IF EXISTS ix_variant_packs_drug_variant_id")
cursor.execute("DROP TABLE variant_packs")
cursor.execute("ALTER TABLE variant_packs_new RENAME TO variant_packs")
cursor.execute("CREATE INDEX ix_variant_packs_drug_variant_id ON variant_packs (drug_variant_id)")
print("Dropped label column from variant_packs")
else:
print("variant_packs.label already absent, skipping")
conn.commit()
finally:
conn.close()
backend/app/models.py
+11 -1
View File
@@ -41,7 +41,6 @@ class VariantPack(Base):
id = Column(Integer, primary_key=True, index=True)
drug_variant_id = Column(Integer, ForeignKey("drug_variants.id"), nullable=False, index=True)
label = Column(String, nullable=False)
pack_unit_name = Column(String, nullable=False, default="pack")
pack_size_in_base_units = Column(Float, nullable=False, default=1)
is_active = Column(Boolean, nullable=False, default=True)
@@ -109,6 +108,17 @@ class DispensingAllocation(Base):
quantity = Column(Float, nullable=False)
class GtinMapping(Base):
__tablename__ = "gtin_mappings"
id = Column(Integer, primary_key=True, index=True)
gtin = Column(String(14), unique=True, index=True, nullable=False)
drug_variant_id = Column(Integer, ForeignKey("drug_variants.id"), nullable=False, index=True)
variant_pack_id = Column(Integer, ForeignKey("variant_packs.id"), nullable=False, index=True)
created_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
class AuditLog(Base):
__tablename__ = "audit_logs"