Hide referrals of soft-deleted organizations

Author: jaytaph

app/db/db.py

@@ -30,6 +30,8 @@ def __init__(self, config_database: ConfigDatabase):
                 )
                 with self.engine.connect() as conn:
                     conn.execute(text("PRAGMA foreign_keys=ON"))
+                    # Postgres schemas are emulated with attached databases
+                    conn.execute(text("ATTACH DATABASE ':memory:' AS admin"))
             else:
                 self.engine = create_engine(
                     dsn,

app/db/models/organization.py

@@ -0,0 +1,21 @@
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import TIMESTAMP, String
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.types import Uuid
+
+from app.db.models.base import Base
+
+
+class OrganizationEntity(Base):
+    """
+    Organizations managed by the nvi-beheer-api. NVI only reads this table.
+    """
+
+    __tablename__ = "organizations"
+    __table_args__ = ({"schema": "admin"},)
+
+    id: Mapped[UUID] = mapped_column("id", Uuid, primary_key=True)
+    external_id: Mapped[str] = mapped_column("external_id", String)
+    deleted_at: Mapped[datetime | None] = mapped_column("deleted_at", TIMESTAMP)

app/db/repository/referral_repository.py

@@ -1,14 +1,27 @@
 from collections.abc import Sequence
 from uuid import UUID
 
-from sqlalchemy import and_, delete, exists, select
+from sqlalchemy import ColumnElement, and_, delete, exists, not_, or_, select
 from sqlalchemy.exc import SQLAlchemyError
 
 from app.db.decorator import repository
+from app.db.models.organization import OrganizationEntity
 from app.db.models.referral import ReferralEntity
 from app.db.repository.respository_base import RepositoryBase
 
 
+def _organization_not_deleted() -> ColumnElement[bool]:
+    """
+    Hide referrals whose organization is soft-deleted, unless an active organization with the same
+    URA number exists (deleted and recreated). Referrals without any organization stay visible.
+    """
+    same_ura = OrganizationEntity.external_id == ReferralEntity.ura_number
+    return or_(
+        exists().where(same_ura, OrganizationEntity.deleted_at.is_(None)),
+        not_(exists().where(same_ura)),
+    )
+
+
 @repository(ReferralEntity)
 class ReferralRepository(RepositoryBase):
     def find_one(self, pseudonym: str, ura_number: str, source: str) -> ReferralEntity | None:
@@ -21,7 +34,7 @@ def find_one(self, pseudonym: str, ura_number: str, source: str) -> ReferralEnti
         return result
 
     def find_by_id(self, id: UUID) -> ReferralEntity | None:
-        stmt = select(ReferralEntity).where(ReferralEntity.id == id)
+        stmt = select(ReferralEntity).where(ReferralEntity.id == id, _organization_not_deleted())
         results = self.db_session.execute(stmt).scalars().first()
         return results
 
@@ -31,7 +44,7 @@ def find_many(
         ura_number: str | None = None,
         source: str | None = None,
     ) -> Sequence[ReferralEntity]:
-        stmt = select(ReferralEntity)
+        stmt = select(ReferralEntity).where(_organization_not_deleted())
 
         if ura_number is not None:
             stmt = stmt.where(ReferralEntity.ura_number == ura_number)

sql/026-admin-schema.sql

@@ -10,6 +10,7 @@ CREATE TABLE admin.organizations (
     CONSTRAINT organizations_pkey PRIMARY KEY (id)
 );
 CREATE UNIQUE INDEX uq_organizations_external_id_active ON admin.organizations (external_id) WHERE deleted_at IS NULL;
+CREATE INDEX ix_organizations_external_id ON admin.organizations (external_id);
 
 CREATE TABLE admin.clients (
     id uuid DEFAULT gen_random_uuid() NOT NULL,

tests/db/repository/test_referral_repository.py

@@ -1,3 +1,4 @@
+from datetime import UTC, datetime
 from uuid import uuid4
 
 import pytest
@@ -7,6 +8,7 @@
 from app.config import ConfigDatabase
 from app.db.db import Database
 from app.db.models.key_info import KeyInfoEntity
+from app.db.models.organization import OrganizationEntity
 from app.db.models.referral import ReferralEntity
 from app.db.repository.referral_repository import ReferralRepository
 
@@ -323,3 +325,54 @@ def test_exists_should_return_false_if_record_is_not_there(
         )
 
         assert exist is False
+
+
+NOW = datetime(2026, 1, 1, tzinfo=UTC)
+
+
+@pytest.mark.parametrize(
+    "organizations,visible",
+    [
+        pytest.param([None], True, id="active"),
+        pytest.param([NOW], False, id="deleted"),
+        pytest.param([NOW, None], True, id="deleted-then-recreated"),
+        pytest.param([], True, id="no-organization"),
+    ],
+)
+def test_referral_visibility_depends_on_organization(
+    referral_repository: ReferralRepository,
+    mock_referral_entity: ReferralEntity,
+    mock_key_info: KeyInfoEntity,
+    organizations: list[datetime | None],
+    visible: bool,
+) -> None:
+    with referral_repository.db_session:
+        for deleted_at in organizations:
+            referral_repository.db_session.add(
+                OrganizationEntity(id=uuid4(), external_id=mock_referral_entity.ura_number, deleted_at=deleted_at)
+            )
+        mock_referral_entity.key_info = mock_key_info
+        referral_repository.add_one(mock_referral_entity)
+
+        by_id = referral_repository.find_by_id(mock_referral_entity.id)
+        many = referral_repository.find_many(ura_number=mock_referral_entity.ura_number)
+
+    assert (by_id is not None) is visible
+    assert len(many) == (1 if visible else 0)
+
+
+def test_deleted_organization_hides_only_its_own_referrals(
+    referral_repository: ReferralRepository,
+    mock_key_info: KeyInfoEntity,
+) -> None:
+    with referral_repository.db_session:
+        referral_repository.db_session.add(OrganizationEntity(id=uuid4(), external_id="00000001", deleted_at=NOW))
+        referral_repository.db_session.add(OrganizationEntity(id=uuid4(), external_id="00000002", deleted_at=None))
+        for ura in ("00000001", "00000002"):
+            referral_repository.add_one(
+                ReferralEntity(ura_number=ura, pseudonym="some-pseudonym", source="Some-Device", key_info=mock_key_info)
+            )
+
+        actual = referral_repository.find_many(pseudonym="some-pseudonym")
+
+    assert [r.ura_number for r in actual] == ["00000002"]

tests/db/test_organization.py

@@ -0,0 +1,21 @@
+from datetime import UTC, datetime
+from uuid import uuid4
+
+from sqlalchemy import select, text
+from sqlalchemy.orm import Session
+
+from app.db.db import Database
+from app.db.models.organization import OrganizationEntity
+
+
+def test_organizations_live_in_admin_schema(database: Database) -> None:
+    with Session(database.engine) as session:
+        session.add(OrganizationEntity(id=uuid4(), external_id="00000123", deleted_at=datetime.now(UTC)))
+        session.commit()
+
+        actual = session.scalars(select(OrganizationEntity)).one()
+        in_admin_schema = session.scalar(text("SELECT count(*) FROM admin.organizations"))
+
+    assert actual.external_id == "00000123"
+    assert actual.deleted_at is not None
+    assert in_admin_schema == 1