"""Job number lookup abstraction — the seam for the future Infor VISUAL ERP integration. Today, Job Number is free text: the default NullJobLookupService accepts any value and returns no enrichment. When PESCO is ready to integrate VISUAL, implement VisualJobLookupService below, set JOB_LOOKUP_PROVIDER=visual (plus the VISUAL_DB_* variables) in .env, and restart — no schema or frontend changes required: * the NCR schema already stores the job number exactly as entered, plus a related `job_info` row (part_id, part_description, customer_name, work_order_status) that any provider can populate at NCR creation; * the frontend job-number field already calls GET /api/jobs/{job_number}/lookup as the user types and displays whatever enrichment comes back, so validation/autocomplete light up automatically with a real provider. """ import logging from dataclasses import dataclass from typing import Protocol from app.config import get_settings logger = logging.getLogger(__name__) @dataclass class JobInfoData: part_id: str | None = None part_description: str | None = None customer_name: str | None = None work_order_status: str | None = None source: str = "null" class JobLookupService(Protocol): async def lookup(self, job_number: str) -> JobInfoData | None: """Return read-only enrichment for a job number, or None when the job is unknown / the provider has nothing to add. Implementations must never raise for a merely-unknown job number.""" ... class NullJobLookupService: """Default provider: job numbers are accepted as-is, no enrichment.""" async def lookup(self, job_number: str) -> JobInfoData | None: # noqa: ARG002 return None class VisualJobLookupService: """PLACEHOLDER for the future Infor VISUAL Manufacturing (SQL Server) integration. Not implemented yet — selecting JOB_LOOKUP_PROVIDER=visual today raises at startup with a pointer here. Implementation notes (verified against PESCO's VISUAL 10 schema): * Connect read-only to the VISUAL SQL Server database (VISUAL_DB_* env vars) with a dedicated SELECT-only SQL login. Use `aioodbc` or `pymssql`. NEVER write to VISUAL tables — hundreds of triggers maintain derived values and direct writes bypass application validation. * A PESCO "job number" corresponds to a work order base id (typically with lot/split/sub qualifiers). WORK_ORDER's primary key is composite: (TYPE, BASE_ID, LOT_ID, SPLIT_ID, SUB_ID); manufacturing work orders have TYPE = 'W'. Parse the entered job number into BASE_ID (and LOT_ID when the shop uses BASE/LOT notation, e.g. "12345/1") and query: SELECT TOP 1 wo.BASE_ID, wo.LOT_ID, wo.SUB_ID, wo.PART_ID, wo.STATUS, wo.DESIRED_QTY, wo.CREATE_DATE, p.DESCRIPTION AS PART_DESCRIPTION FROM WORK_ORDER wo LEFT JOIN PART p ON p.ID = wo.PART_ID WHERE wo.TYPE = 'W' AND wo.BASE_ID = :base_id ORDER BY wo.LOT_ID, wo.SPLIT_ID, wo.SUB_ID STATUS is a one-char code (R=released, C=closed, etc.) — map it to a readable label for work_order_status. * Customer enrichment goes through the demand/supply linkage: DEMAND_SUPPLY_LINK rows with SUPPLY_TYPE='WO' and SUPPLY_BASE_ID = wo.BASE_ID (match SUPPLY_LOT_ID/SUPPLY_SPLIT_ID/SUPPLY_SUB_ID when present) point at customer-order demand (DEMAND_TYPE='CO', DEMAND_BASE_ID = CUST_ORDER_LINE.CUST_ORDER_ID, DEMAND_SEQ_NO = line no). Join CUSTOMER_ORDER -> CUSTOMER for the customer name. * Return JobInfoData(part_id=..., part_description=..., customer_name=..., work_order_status=..., source="visual"). Return None when no WORK_ORDER row matches. Wrap connection errors in logging + return None so an ERP outage never blocks NCR entry. """ def __init__(self) -> None: settings = get_settings() raise NotImplementedError( "VisualJobLookupService is a documented stub. Implement it per the " "notes in app/services/job_lookup.py, or set JOB_LOOKUP_PROVIDER=null. " f"(Configured VISUAL host: {settings.visual_db_host or 'unset'})" ) async def lookup(self, job_number: str) -> JobInfoData | None: raise NotImplementedError _service: JobLookupService | None = None def get_job_lookup_service() -> JobLookupService: global _service if _service is None: provider = get_settings().job_lookup_provider if provider == "visual": _service = VisualJobLookupService() # raises: intentionally loud else: _service = NullJobLookupService() logger.info("Job lookup provider: %s", provider) return _service