156 lines
4.3 KiB
Python
156 lines
4.3 KiB
Python
"""Low-level DB access for the barcode WMS client.
|
|
|
|
The goal of this module is to mirror the legacy C# barcode form as closely as
|
|
possible while keeping SQL isolated from the UI.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
SQL_NEXT_PICKING = """
|
|
SELECT TOP (1)
|
|
Documento,
|
|
CodNazione,
|
|
NAZIONE,
|
|
Stato,
|
|
Pallet,
|
|
Cella,
|
|
Ubicazione,
|
|
Ordinamento,
|
|
IDStato
|
|
FROM dbo.py_XMag_ViewPackingList
|
|
WHERE Ordinamento > 0
|
|
AND IDStato = :id_stato
|
|
ORDER BY Ordinamento;
|
|
"""
|
|
|
|
SQL_PICKING_BY_PALLET = """
|
|
SELECT TOP (1)
|
|
Documento,
|
|
CodNazione,
|
|
NAZIONE,
|
|
Stato,
|
|
Pallet,
|
|
Cella,
|
|
Ubicazione,
|
|
Ordinamento,
|
|
IDStato
|
|
FROM dbo.py_XMag_ViewPackingList
|
|
WHERE Pallet = :pallet
|
|
ORDER BY Ordinamento;
|
|
"""
|
|
|
|
SQL_TRACE_BY_PALLET = """
|
|
SELECT TOP (1)
|
|
Pallet,
|
|
Lotto,
|
|
Prodotto,
|
|
Descrizione
|
|
FROM dbo.vXTracciaProdotti
|
|
WHERE Pallet = :pallet
|
|
ORDER BY Lotto;
|
|
"""
|
|
|
|
SQL_LEGACY_MOVE = """
|
|
SET NOCOUNT ON;
|
|
DECLARE @RC int = 0;
|
|
|
|
EXEC dbo.sp_xMagGestioneMagazziniPallet
|
|
@IDOperatore = :id_operatore,
|
|
@BarcodeCella = :barcode_cella,
|
|
@BarcodePallet = :barcode_pallet,
|
|
@NumeroCella = :numero_cella,
|
|
@RC = @RC OUTPUT;
|
|
|
|
EXEC dbo.py_sp_ControllaPrenotazionePackingListPalletNew;
|
|
|
|
SELECT
|
|
@RC AS RC,
|
|
:barcode_cella AS BarcodeCella,
|
|
:barcode_pallet AS BarcodePallet,
|
|
:numero_cella AS NumeroCella;
|
|
"""
|
|
|
|
|
|
def _rows_to_dicts(res: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
"""Convert ``query_json`` payloads to a list of row dictionaries."""
|
|
|
|
if not isinstance(res, dict):
|
|
return []
|
|
rows = res.get("rows") or []
|
|
cols = res.get("columns") or []
|
|
if rows and isinstance(rows[0], dict):
|
|
return [row for row in rows if isinstance(row, dict)]
|
|
out: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
if isinstance(row, (list, tuple)) and cols:
|
|
out.append({str(cols[i]): row[i] for i in range(min(len(cols), len(row)))})
|
|
return out
|
|
|
|
|
|
@dataclass
|
|
class LegacyMoveResult:
|
|
"""Result of one movement executed through the legacy stored procedure."""
|
|
|
|
rc: int
|
|
barcode_cella: str
|
|
barcode_pallet: str
|
|
numero_cella: int
|
|
|
|
|
|
class BarcodeRepository:
|
|
"""Thin async repository used by the lightweight barcode client."""
|
|
|
|
def __init__(self, db_client):
|
|
self.db_client = db_client
|
|
|
|
async def fetch_next_picking(self, id_stato: int) -> dict[str, Any] | None:
|
|
"""Return the next pallet proposed by the legacy F1/F2 queue logic."""
|
|
|
|
res = await self.db_client.query_json(SQL_NEXT_PICKING, {"id_stato": int(id_stato)})
|
|
rows = _rows_to_dicts(res)
|
|
return rows[0] if rows else None
|
|
|
|
async def fetch_picking_by_pallet(self, pallet: str) -> dict[str, Any] | None:
|
|
"""Return one picking row for the given pallet, if still present in the queue."""
|
|
|
|
res = await self.db_client.query_json(SQL_PICKING_BY_PALLET, {"pallet": str(pallet or "").strip()})
|
|
rows = _rows_to_dicts(res)
|
|
return rows[0] if rows else None
|
|
|
|
async def fetch_trace_by_pallet(self, pallet: str) -> dict[str, Any] | None:
|
|
"""Return trace information used by the C# fallback confirmation path."""
|
|
|
|
res = await self.db_client.query_json(SQL_TRACE_BY_PALLET, {"pallet": str(pallet or "").strip()})
|
|
rows = _rows_to_dicts(res)
|
|
return rows[0] if rows else None
|
|
|
|
async def execute_legacy_move(
|
|
self,
|
|
*,
|
|
operator_id: int,
|
|
barcode_cella: str,
|
|
barcode_pallet: str,
|
|
numero_cella: int,
|
|
) -> LegacyMoveResult:
|
|
"""Execute the same stored procedure used by the C# barcode form."""
|
|
|
|
params = {
|
|
"id_operatore": int(operator_id),
|
|
"barcode_cella": str(barcode_cella or "").strip(),
|
|
"barcode_pallet": str(barcode_pallet or "").strip(),
|
|
"numero_cella": int(numero_cella),
|
|
}
|
|
res = await self.db_client.query_json(SQL_LEGACY_MOVE, params, commit=True)
|
|
rows = _rows_to_dicts(res)
|
|
row = rows[0] if rows else {}
|
|
return LegacyMoveResult(
|
|
rc=int(row.get("RC") or 0),
|
|
barcode_cella=str(row.get("BarcodeCella") or params["barcode_cella"]),
|
|
barcode_pallet=str(row.get("BarcodePallet") or params["barcode_pallet"]),
|
|
numero_cella=int(row.get("NumeroCella") or params["numero_cella"]),
|
|
)
|