feat(integrations): LabelData schema + Snipe-IT lookup client

backend/app/schemas/__init__.py

@@ -0,0 +1 @@
+"""Pydantic schemas — Request/Response/Domain value-objects."""

backend/app/schemas/label_data.py

@@ -0,0 +1,28 @@
+"""App-agnostic label data passed from lookup-clients to the LabelRenderer.
+
+LabelData is what a `*_client.lookup(id)` call produces. It is the
+serialisable view of a real-world entity (Snipe-IT asset, Grocy product,
+Spoolman spool) condensed into the minimal set of fields a label needs:
+a title, an identifier to print, a QR-encodable URL, optional secondary
+lines, and a source-app tag for downstream routing.
+
+Layout, font, geometry, and tape-fit decisions live on the LabelRenderer
+side — they are NOT in this model.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict
+
+
+class LabelData(BaseModel):
+    """Immutable, app-agnostic label payload."""
+
+    model_config = ConfigDict(frozen=True)
+
+    title: str
+    primary_id: str
+    qr_payload: str
+    source_app: str
+    secondary: tuple[str, ...] = ()
+    """Additional label lines below the primary identifier."""

backend/app/services/snipeit_client.py

@@ -0,0 +1,79 @@
+"""Snipe-IT REST API client — asset lookup by asset_tag.
+
+The client emits domain-level `LabelData` records so downstream consumers
+(LabelRenderer, queue submitters) never see Snipe-IT's raw schema. Add new
+fields by extending the mapping in `lookup()`, never by leaking the upstream
+shape.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import quote
+
+import httpx
+
+from app.schemas.label_data import LabelData
+
+
+class SnipeITNotFoundError(Exception):
+    """Raised when no Snipe-IT asset matches the given tag."""
+
+
+class SnipeITClient:
+    """Async client for Snipe-IT's REST API.
+
+    Authenticates with a bearer token (Snipe-IT API key). Configuration —
+    base URL, API key, timeout — is injected so the same class can hit the
+    user's live instance from production and a respx-mocked endpoint from
+    tests, with no hidden global state.
+    """
+
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        api_key: str,
+        timeout: float = 5.0,
+    ) -> None:
+        self._base_url = base_url.rstrip("/")
+        self._api_key = api_key
+        self._timeout = timeout
+
+    async def lookup(self, asset_tag: str) -> LabelData:
+        """Return LabelData for `asset_tag`, or raise SnipeITNotFoundError."""
+        # TODO(phase6): inject a shared httpx.AsyncClient for connection pooling
+        #               when this client is consumed by the FastAPI request handler.
+        encoded_tag = quote(asset_tag, safe="")
+        url = f"{self._base_url}/api/v1/hardware/bytag/{encoded_tag}"
+        headers = {
+            "Authorization": f"Bearer {self._api_key}",
+            "Accept": "application/json",
+        }
+        async with httpx.AsyncClient(timeout=self._timeout) as client:
+            response = await client.get(url, headers=headers)
+
+        if response.status_code == 404:
+            raise SnipeITNotFoundError(f"Asset {asset_tag!r} not found")
+        # 401/403/5xx surface as httpx.HTTPStatusError — callers (AppLookupService)
+        # decide whether to treat them as configuration errors vs transient failures.
+        response.raise_for_status()
+
+        payload: dict[str, Any] = response.json()
+        return self._payload_to_label(payload, asset_tag)
+
+    def _payload_to_label(self, payload: dict[str, Any], asset_tag: str) -> LabelData:
+        asset_id = payload.get("id")
+        if asset_id is None:
+            raise ValueError(f"Snipe-IT response for {asset_tag!r} is missing required field 'id'")
+        secondary: list[str] = []
+        serial = payload.get("serial")
+        if serial:
+            secondary.append(f"S/N: {serial}")
+        return LabelData(
+            title=str(payload.get("name") or asset_tag),
+            primary_id=str(payload.get("asset_tag") or asset_tag),
+            qr_payload=f"{self._base_url}/hardware/{asset_id}",
+            source_app="snipeit",
+            secondary=tuple(secondary),
+        )

backend/tests/unit/schemas/test_label_data.py

@@ -0,0 +1,56 @@
+import pytest
+from app.schemas.label_data import LabelData
+from pydantic_core import ValidationError
+
+
+def test_label_data_minimal() -> None:
+    data = LabelData(
+        title="MacBook Pro 16",
+        primary_id="ASSET-12345",
+        qr_payload="https://snipe-it.example/assets/12345",
+        source_app="snipeit",
+    )
+    assert data.title == "MacBook Pro 16"
+    assert data.primary_id == "ASSET-12345"
+    assert data.qr_payload == "https://snipe-it.example/assets/12345"
+    assert data.source_app == "snipeit"
+    assert data.secondary == ()
+
+
+def test_label_data_with_secondary_fields() -> None:
+    data = LabelData(
+        title="BambuLab PLA",
+        primary_id="#42",
+        qr_payload="https://spoolman.example/spool/42",
+        source_app="spoolman",
+        secondary=["Color: Black", "Weight: 850g"],
+    )
+    assert len(data.secondary) == 2
+    assert data.secondary[0] == "Color: Black"
+    # The tuple is the actual immutability guarantee — confirm.
+    assert isinstance(data.secondary, tuple)
+
+
+def test_label_data_is_frozen() -> None:
+    """LabelData is an immutable value object — mutating fields after construction must fail."""
+    data = LabelData(
+        title="t",
+        primary_id="p",
+        qr_payload="q",
+        source_app="snipeit",
+    )
+    with pytest.raises(ValidationError, match="frozen_instance"):
+        data.title = "different"  # type: ignore[misc]
+
+
+def test_label_data_secondary_is_immutable() -> None:
+    """A tuple field cannot be mutated in-place — append must raise AttributeError."""
+    data = LabelData(
+        title="t",
+        primary_id="p",
+        qr_payload="q",
+        source_app="snipeit",
+        secondary=["a"],
+    )
+    with pytest.raises(AttributeError):
+        data.secondary.append("b")  # type: ignore[attr-defined]

backend/tests/unit/services/test_snipeit_client.py

@@ -0,0 +1,132 @@
+import httpx
+import pytest
+import respx
+from app.services.snipeit_client import SnipeITClient, SnipeITNotFoundError
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_asset_returns_label_data() -> None:
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/ASSET-12345").mock(
+        return_value=httpx.Response(
+            200,
+            json={
+                "id": 123,
+                "asset_tag": "ASSET-12345",
+                "name": "MacBook Pro 16",
+                "serial": "C02XYZ",
+            },
+        )
+    )
+
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key")
+    data = await client.lookup("ASSET-12345")
+
+    assert data.title == "MacBook Pro 16"
+    assert data.primary_id == "ASSET-12345"
+    assert data.qr_payload == "https://snipe-it.example/hardware/123"
+    assert data.source_app == "snipeit"
+    assert data.secondary == ("S/N: C02XYZ",)
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_asset_404_raises_not_found() -> None:
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/UNKNOWN").mock(
+        return_value=httpx.Response(404)
+    )
+
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key")
+
+    with pytest.raises(SnipeITNotFoundError, match="UNKNOWN"):
+        await client.lookup("UNKNOWN")
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_asset_without_serial_has_no_secondary_line() -> None:
+    """Missing optional serial field must not add a 'S/N: None' line."""
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock(
+        return_value=httpx.Response(
+            200,
+            json={"id": 1, "asset_tag": "A-1", "name": "Thing"},
+        )
+    )
+
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key")
+    data = await client.lookup("A-1")
+
+    assert data.secondary == ()
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_strips_trailing_slash_from_base_url() -> None:
+    """base_url='https://snipe-it.example/' must not produce a double slash."""
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock(
+        return_value=httpx.Response(200, json={"id": 1, "asset_tag": "A-1", "name": "Thing"})
+    )
+    client = SnipeITClient(base_url="https://snipe-it.example/", api_key="test-key")
+    data = await client.lookup("A-1")
+    assert data.qr_payload == "https://snipe-it.example/hardware/1"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_missing_id_raises_value_error() -> None:
+    """Snipe-IT response without 'id' field must fail loudly.
+
+    Regression guard: must not silently produce …/hardware/None.
+    """
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock(
+        return_value=httpx.Response(
+            200,
+            json={"asset_tag": "A-1", "name": "Broken Asset"},  # no 'id'
+        )
+    )
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key")
+    with pytest.raises(ValueError, match="missing required field 'id'"):
+        await client.lookup("A-1")
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_5xx_raises_httpx_error() -> None:
+    """A 500 from upstream must surface as httpx.HTTPStatusError (no swallowing)."""
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock(
+        return_value=httpx.Response(500)
+    )
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key")
+    with pytest.raises(httpx.HTTPStatusError):
+        await client.lookup("A-1")
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_url_encodes_asset_tag() -> None:
+    """Asset tags with special chars (/, ?, space) must be percent-encoded."""
+    respx.get("https://snipe-it.example/api/v1/hardware/bytag/A%2F1%20test").mock(
+        return_value=httpx.Response(
+            200,
+            json={"id": 1, "asset_tag": "A/1 test", "name": "Thing"},
+        )
+    )
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="test-key")
+    data = await client.lookup("A/1 test")
+    assert data.title == "Thing"
+
+
+@pytest.mark.asyncio
+@respx.mock
+async def test_lookup_sends_bearer_auth_header() -> None:
+    """lookup() must send Authorization: Bearer … and Accept: application/json."""
+    route = respx.get("https://snipe-it.example/api/v1/hardware/bytag/A-1").mock(
+        return_value=httpx.Response(200, json={"id": 1, "asset_tag": "A-1", "name": "T"})
+    )
+    client = SnipeITClient(base_url="https://snipe-it.example", api_key="secret-key-42")
+    await client.lookup("A-1")
+
+    assert route.called
+    sent_request = route.calls.last.request
+    assert sent_request.headers["Authorization"] == "Bearer secret-key-42"
+    assert sent_request.headers["Accept"] == "application/json"