pre-commit fixed

Author: shaikhhasib

alembic/versions/2026_03_10_0000-6c1c0c2f8a1b_adjust_pv_uniqueness.py

@@ -0,0 +1,31 @@
+"""Adjust PV address uniqueness constraints.
+
+Revision ID: 6c1c0c2f8a1b
+Revises: 61476c608fc3
+Create Date: 2026-03-10 00:00:00.000000
+"""
+from collections.abc import Sequence
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "6c1c0c2f8a1b"
+down_revision: str | None = "61476c608fc3"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.drop_index(op.f("ix_pv_readback_address"), table_name="pv")
+    op.drop_index(op.f("ix_pv_config_address"), table_name="pv")
+
+    op.create_index(op.f("ix_pv_readback_address"), "pv", ["readback_address"], unique=False)
+    op.create_index(op.f("ix_pv_config_address"), "pv", ["config_address"], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_pv_readback_address"), table_name="pv")
+    op.drop_index(op.f("ix_pv_config_address"), table_name="pv")
+
+    op.create_index(op.f("ix_pv_readback_address"), "pv", ["readback_address"], unique=True)
+    op.create_index(op.f("ix_pv_config_address"), "pv", ["config_address"], unique=True)

app/models/pv.py

@@ -26,8 +26,8 @@ class PV(Base, UUIDMixin, TimestampMixin):
 
     # PV addresses (at least one required - enforced at service layer)
     setpoint_address: Mapped[str | None] = mapped_column(String(255), unique=True, nullable=True, index=True)
-    readback_address: Mapped[str | None] = mapped_column(String(255), unique=True, nullable=True, index=True)
-    config_address: Mapped[str | None] = mapped_column(String(255), unique=True, nullable=True, index=True)
+    readback_address: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    config_address: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
 
     # Metadata
     device: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)

app/repositories/pv_repository.py

@@ -22,6 +22,28 @@ async def find_by_address(self, address: str) -> PV | None:
         )
         return result.scalar_one_or_none()
 
+    async def find_by_setpoint(self, setpoint_address: str) -> PV | None:
+        """Find PV by setpoint address only."""
+        result = await self.session.execute(
+            select(PV)
+            .options(selectinload(PV.tags).selectinload(Tag.group))
+            .where(PV.setpoint_address == setpoint_address)
+        )
+        return result.scalar_one_or_none()
+
+    async def get_existing_setpoints(self, setpoint_addresses: list[str]) -> set[str]:
+        """Return set of setpoint addresses that already exist."""
+        if not setpoint_addresses:
+            return set()
+        # PostgreSQL limit is 32767 parameters, use a safe batch size.
+        batch_size = 30000
+        existing: set[str] = set()
+        for i in range(0, len(setpoint_addresses), batch_size):
+            batch = setpoint_addresses[i : i + batch_size]
+            result = await self.session.execute(select(PV.setpoint_address).where(PV.setpoint_address.in_(batch)))
+            existing.update({row[0] for row in result.all() if row[0]})
+        return existing
+
     async def search_by_name(
         self,
         search: str | None = None,

app/services/pv_service.py

@@ -1,3 +1,4 @@
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.pv import PV
@@ -33,6 +34,14 @@ def _to_dto(self, pv: PV) -> PVElementDTO:
             lastModifiedDate=pv.updated_at,
         )
 
+    @staticmethod
+    def _normalize_address(address: str | None) -> str | None:
+        """Normalize optional PV addresses from API payloads."""
+        if address is None:
+            return None
+        normalized = address.strip()
+        return normalized or None
+
     async def search_paged(
         self,
         search: str | None = None,
@@ -61,12 +70,17 @@ async def get_by_id(self, pv_id: str) -> PVElementDTO | None:
 
     async def create(self, data: NewPVElementDTO) -> PVElementDTO:
         """Create a new PV."""
+        setpoint_address = self._normalize_address(data.setpointAddress)
+        readback_address = self._normalize_address(data.readbackAddress)
+        config_address = self._normalize_address(data.configAddress)
+        if not any([setpoint_address, readback_address, config_address]):
+            raise ValueError("At least one address (setpoint, readback, or config) is required")
+
         # Check for duplicate addresses
-        for address in [data.setpointAddress, data.readbackAddress, data.configAddress]:
-            if address:
-                existing = await self.pv_repo.find_by_address(address)
-                if existing:
-                    raise ValueError(f"PV with address '{address}' already exists")
+        if setpoint_address:
+            existing = await self.pv_repo.find_by_setpoint(setpoint_address)
+            if existing:
+                raise ValueError(f"PV with setpoint address '{setpoint_address}' already exists")
 
         # Get tags
         tags = []
@@ -76,9 +90,9 @@ async def create(self, data: NewPVElementDTO) -> PVElementDTO:
                 raise ValueError("One or more tag IDs are invalid")
 
         pv = PV(
-            setpoint_address=data.setpointAddress,
-            readback_address=data.readbackAddress,
-            config_address=data.configAddress,
+            setpoint_address=setpoint_address,
+            readback_address=readback_address,
+            config_address=config_address,
             device=data.device,
             description=data.description,
             abs_tolerance=data.absTolerance,
@@ -87,14 +101,41 @@ async def create(self, data: NewPVElementDTO) -> PVElementDTO:
             tags=tags,
         )
 
-        pv = await self.pv_repo.create(pv)
+        try:
+            pv = await self.pv_repo.create(pv)
+        except IntegrityError as e:
+            raise ValueError(f"PV addresses violate uniqueness constraints: {e}") from e
         return self._to_dto(pv)
 
     async def create_many(self, data_list: list[NewPVElementDTO]) -> list[PVElementDTO]:
         """Bulk create PVs."""
+        normalized_records: list[tuple[str | None, str | None, str | None, NewPVElementDTO]] = []
+        for data in data_list:
+            setpoint_address = self._normalize_address(data.setpointAddress)
+            readback_address = self._normalize_address(data.readbackAddress)
+            config_address = self._normalize_address(data.configAddress)
+            if not any([setpoint_address, readback_address, config_address]):
+                raise ValueError("At least one address (setpoint, readback, or config) is required")
+            normalized_records.append((setpoint_address, readback_address, config_address, data))
+
+        setpoints = [r[0] for r in normalized_records if r[0]]
+        seen = set()
+        duplicate_setpoints = set()
+        for s in setpoints:
+            if s in seen:
+                duplicate_setpoints.add(s)
+            else:
+                seen.add(s)
+        if duplicate_setpoints:
+            raise ValueError(f"Duplicate setpoint addresses in import: {sorted(duplicate_setpoints)[:10]}")
+
+        existing_setpoints = await self.pv_repo.get_existing_setpoints(setpoints)
+        if existing_setpoints:
+            raise ValueError(f"Setpoint addresses already exist: {sorted(existing_setpoints)[:10]}")
+
         # Collect all tag IDs
         all_tag_ids = set()
-        for data in data_list:
+        for _, _, _, data in normalized_records:
             all_tag_ids.update(data.tags)
 
         # Fetch all tags at once
@@ -105,12 +146,12 @@ async def create_many(self, data_list: list[NewPVElementDTO]) -> list[PVElementD
 
         # Create PV objects
         pvs = []
-        for data in data_list:
+        for setpoint_address, readback_address, config_address, data in normalized_records:
             pv_tags = [tags_by_id[tid] for tid in data.tags if tid in tags_by_id]
             pv = PV(
-                setpoint_address=data.setpointAddress,
-                readback_address=data.readbackAddress,
-                config_address=data.configAddress,
+                setpoint_address=setpoint_address,
+                readback_address=readback_address,
+                config_address=config_address,
                 device=data.device,
                 description=data.description,
                 abs_tolerance=data.absTolerance,
@@ -121,7 +162,10 @@ async def create_many(self, data_list: list[NewPVElementDTO]) -> list[PVElementD
             pvs.append(pv)
 
         # Bulk insert
-        pvs = await self.pv_repo.bulk_create(pvs)
+        try:
+            pvs = await self.pv_repo.bulk_create(pvs)
+        except IntegrityError as e:
+            raise ValueError(f"PV addresses violate uniqueness constraints: {e}") from e
         return [self._to_dto(pv) for pv in pvs]
 
     async def update(self, pv_id: str, data: UpdatePVElementDTO) -> PVElementDTO | None:

tests/test_api/test_pvs.py

@@ -110,6 +110,40 @@ async def test_bulk_create_empty_list(self, client: AsyncClient):
         data = response.json()
         assert data["payload"] == []
 
+    @pytest.mark.asyncio
+    async def test_bulk_create_allows_blank_optional_addresses(self, client: AsyncClient):
+        """Blank optional addresses from CSV imports should be treated as NULL."""
+        pvs_data = [
+            {"setpointAddress": "CSV:PV:1", "readbackAddress": "", "configAddress": "", "device": "CSV-1"},
+            {"setpointAddress": "CSV:PV:2", "readbackAddress": "", "configAddress": "", "device": "CSV-2"},
+        ]
+
+        response = await client.post("/v1/pvs/multi", json=pvs_data)
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["errorCode"] == 0
+        assert len(data["payload"]) == 2
+        assert data["payload"][0]["readbackAddress"] is None
+        assert data["payload"][0]["configAddress"] is None
+        assert data["payload"][1]["readbackAddress"] is None
+        assert data["payload"][1]["configAddress"] is None
+
+    @pytest.mark.asyncio
+    async def test_bulk_create_allows_duplicate_readback(self, client: AsyncClient):
+        """Readback addresses can repeat across different setpoints."""
+        pvs_data = [
+            {"setpointAddress": "CSV:PV:3", "readbackAddress": "DUP:RB:1", "device": "CSV-3"},
+            {"setpointAddress": "CSV:PV:4", "readbackAddress": "DUP:RB:1", "device": "CSV-4"},
+        ]
+
+        response = await client.post("/v1/pvs/multi", json=pvs_data)
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["errorCode"] == 0
+        assert len(data["payload"]) == 2
+
 
 class TestPVSearch:
     """Tests for PV search endpoints."""