Merge pull request #321 from minvws/cleanup-unused-oin-pattern

Refactor to use OinType

Author: ricklambrechts

app/db/entities/hsm_key_versions.py

@@ -8,6 +8,7 @@
 
 from app.db.entities.base import Base
 from app.db.entities.organization import Organization
+from app.models.oin import Oin
 
 
 class HsmKeyVersion(Base):
@@ -37,13 +38,13 @@ class HsmKeyVersion(Base):
     organization: Mapped[Organization] = relationship("Organization")
 
     @property
-    def oin(self) -> str:
+    def oin(self) -> Oin:
         return self.organization.oin
 
     def to_dict(self) -> dict[str, Any]:
         return {
             "id": str(self.id),
-            "oin": self.oin,
+            "oin": self.oin.value,
             "version": self.version,
             "from_dt": self.from_dt,
             "until_dt": self.until_dt,

app/db/entities/organization.py

@@ -1,6 +1,8 @@
 import uuid
 from typing import Any, List, TYPE_CHECKING, Optional
 
+from app.db.types.oin import OinType
+from app.models.oin import Oin
 from sqlalchemy import String
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -21,7 +23,7 @@ class Organization(Base):
     id: Mapped[uuid.UUID] = mapped_column(
         UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
     )
-    oin: Mapped[str] = mapped_column("oin", String, unique=True)
+    oin: Mapped[Oin] = mapped_column("oin", OinType(), unique=True)
     name: Mapped[str] = mapped_column("name", String)
     max_rid_usage: Mapped[str] = mapped_column("max_rid_usage", String)
 
@@ -33,7 +35,7 @@ class Organization(Base):
 
     def to_dict(self) -> dict[str, Any]:
         return {
-            "oin": self.oin,
+            "oin": self.oin.value,
             "name": self.name,
             "max_rid_usage": self.max_rid_usage,
         }

app/db/repositories/hsm_key_version_repository.py

@@ -36,7 +36,7 @@ def get_active_versions(
             )
         )
         if oin is not None:
-            query = query.join(Organization).where(Organization.oin == oin.value)
+            query = query.join(Organization).where(Organization.oin == oin)
         return self.db_session.execute(query).scalars().all()
 
     def get_by_oin(self, oin: Oin) -> Sequence[HsmKeyVersion]:
@@ -48,7 +48,7 @@ def get_by_oin(self, oin: Oin) -> Sequence[HsmKeyVersion]:
             select(HsmKeyVersion)
             .options(joinedload(HsmKeyVersion.organization))
             .join(Organization)
-            .where(Organization.oin == oin.value)
+            .where(Organization.oin == oin)
             .order_by(HsmKeyVersion.version)
         )
         return self.db_session.execute(query).scalars().all()

app/db/repositories/org_repository.py

@@ -7,19 +7,21 @@
 from app.db.repositories.repository_base import RepositoryBase
 from sqlalchemy import select
 
+from app.models.oin import Oin
+
 logger = logging.getLogger(__name__)
 
 
 @repository(Organization)
 class OrgRepository(RepositoryBase):
-    def get_by_oin(self, oin: str) -> Optional[Organization]:
+    def get_by_oin(self, oin: Oin) -> Optional[Organization]:
         """
         Fetches the organization by its unique ID.
         """
         query = select(Organization).where(Organization.oin == oin)
         return self.db_session.execute(query).scalars().first()
 
-    def create(self, oin: str, name: str, max_usage_level: str) -> Organization:
+    def create(self, oin: Oin, name: str, max_usage_level: str) -> Organization:
         """
         Creates a new org entry.
         """
@@ -31,7 +33,7 @@ def create(self, oin: str, name: str, max_usage_level: str) -> Organization:
         self.db_session.add(entry)
         self.db_session.flush()
 
-        logger.info("created organization with OIN %s and name %r", oin, name)
+        logger.info("created organization with OIN %s and name %r", oin.value, name)
         return entry
 
     def update(

app/db/types/__init__.py

@@ -0,0 +1,29 @@
+from typing import Any
+
+from sqlalchemy import String
+from sqlalchemy.types import TypeDecorator
+
+from app.models.oin import Oin
+
+
+class OinType(TypeDecorator[Oin]):
+    """Map ``Organization.oin`` to the OIN value object and store it as text."""
+
+    impl = String
+    cache_ok = True
+    python_type = Oin
+
+    def process_bind_param(self, value: Any, _dialect: Any) -> str | None:
+        if value is None:
+            return None
+
+        if isinstance(value, Oin):
+            return value.value
+
+        return str(value)
+
+    def process_result_value(self, value: Any, _dialect: Any) -> Oin | None:
+        if value is None:
+            return None
+
+        return Oin(value)

app/db/types/oin.py

@@ -3,11 +3,18 @@
 
 from pydantic import GetCoreSchemaHandler
 from pydantic.json_schema import JsonSchemaValue
-from pydantic_core import core_schema
+from pydantic_core import PydanticCustomError, core_schema
 
 # An OIN is always 20 characters:
 #   prefix (8 digits) + mainnumber (8 or 9 alphanumeric) + suffix (4 or 3 zeros)
 OIN_PATTERN = re.compile(r"^\d{8}(?:[A-Za-z0-9]{8}0{4}|[A-Za-z0-9]{9}0{3})$")
+RECIPIENT_ORGANIZATION_PREFIX = "oin:"
+_RECIPIENT_ORGANIZATION_SUFFIX_PATTERN = (
+    r"\d{8}(?:[A-Za-z0-9]{8}0{4}|[A-Za-z0-9]{9}0{3})"
+)
+RECIPIENT_ORGANIZATION_PATTERN = re.compile(
+    rf"^{RECIPIENT_ORGANIZATION_PREFIX}{_RECIPIENT_ORGANIZATION_SUFFIX_PATTERN}$"
+)
 
 
 class Oin:
@@ -92,6 +99,8 @@ def __repr__(self) -> str:
     def __eq__(self, other: Any) -> bool:
         if isinstance(other, Oin):
             return self.value == other.value
+        if isinstance(other, str):
+            return self.value == other
         return False
 
     def __hash__(self) -> int:
@@ -128,3 +137,75 @@ def __get_pydantic_json_schema__(
                 "8-digit prefix + 8/9-character alphanumeric mainnumber + 4/3 trailing zeros (suffix)."
             ),
         }
+
+
+class RecipientOrganizationOin(Oin):
+    """An OIN supplied with recipient organization prefix, e.g. ``oin:<oin>``.
+
+    Internally the stored value remains a plain OIN without the ``oin:`` prefix.
+    """
+
+    PREFIX = RECIPIENT_ORGANIZATION_PREFIX
+    _INVALID_ERROR = "Invalid recipient organization. Format: oin:<oin_number>"
+
+    def __init__(self, value: Any) -> None:
+        if isinstance(value, str) and value.startswith(self.PREFIX):
+            value = value[len(self.PREFIX) :]
+        super().__init__(value)
+
+    @classmethod
+    def _pydantic_validate(cls, value: Any) -> "RecipientOrganizationOin":
+        if isinstance(value, cls):
+            return value
+
+        if isinstance(value, Oin):
+            return cls(value.value)
+
+        if not isinstance(value, str):
+            raise PydanticCustomError(
+                "invalid_recipient_organization", cls._INVALID_ERROR
+            )
+
+        if not value.startswith(cls.PREFIX):
+            raise PydanticCustomError(
+                "invalid_recipient_organization", cls._INVALID_ERROR
+            )
+
+        try:
+            return cls(value[len(cls.PREFIX) :])
+        except ValueError as e:
+            raise PydanticCustomError(
+                "invalid_recipient_organization", "{error}", {"error": str(e)}
+            ) from e
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, _source_type: Any, _handler: GetCoreSchemaHandler
+    ) -> core_schema.CoreSchema:
+        return core_schema.no_info_plain_validator_function(
+            cls._pydantic_validate,
+            serialization=core_schema.to_string_ser_schema(),
+        )
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, _schema: core_schema.CoreSchema, _handler: Any
+    ) -> JsonSchemaValue:
+        return {
+            "type": "string",
+            "pattern": RECIPIENT_ORGANIZATION_PATTERN.pattern,
+            "examples": [f"{RECIPIENT_ORGANIZATION_PREFIX}00000099000000001000"],
+            "description": (
+                "Recipient organization string: prefix followed by an OIN (20-char "
+                "identifier)."
+            ),
+        }
+
+    def __str__(self) -> str:
+        """Return the recipient organization OIN in external format, including ``oin:``."""
+        return f"{self.PREFIX}{self.value}"
+
+    @property
+    def value(self) -> str:
+        """Return the normalized OIN value without the ``oin:`` prefix."""
+        return super().value

app/models/oin.py

@@ -1,43 +1,34 @@
 import base64
 import logging
 from datetime import datetime
-from typing import Any, Literal, List, Optional
+from typing import Any, Literal, List
 
 from pydantic import BaseModel, ConfigDict, model_validator, Field, field_validator
 
-from app.models.oin import Oin
+from app.models.oin import Oin, RecipientOrganizationOin
 from app.personal_id import PersonalId
 from app.services.pseudonym_service import PseudonymType
 from app.rid import RidUsage
 
 logger = logging.getLogger(__name__)
-OIN_PATTERN = r"^\d{8}(?:[A-Za-z0-9]{8}0{4}|[A-Za-z0-9]{9}0{3})$"
 
 
 class RegisterRequest(BaseModel):
     scope: List[str]
-    key_id: Optional[str]
+    key_id: str | None
 
 
 class OrgRequest(BaseModel):
-    oin: str
+    oin: Oin
     name: str = Field(..., min_length=5, max_length=50)
     max_key_usage: RidUsage
 
-    @field_validator("oin")
-    def validate_oin(cls, v: Any) -> str:
-        return str(Oin(v))
-
 
 class HsmKeyVersionRequest(BaseModel):
-    oin: str
+    oin: Oin
     from_dt: datetime | None = None
     until_dt: datetime | None = None
 
-    @field_validator("oin")
-    def validate_oin(cls, v: Any) -> str:
-        return str(Oin(v))
-
 
 class HsmKeyVersionUpdateRequest(BaseModel):
     until_dt: datetime | None = None
@@ -46,14 +37,14 @@ class HsmKeyVersionUpdateRequest(BaseModel):
 
 class RidReceiveRequest(BaseModel):
     rid: str
-    recipientOrganization: str
+    recipientOrganization: RecipientOrganizationOin
     recipientScope: str
     pseudonymType: Literal["rp", "irp", "bsn"]
 
 
 class BlindRequest(BaseModel):
     encryptedPersonalId: str = Field(..., min_length=2)
-    recipientOrganization: str = Field(..., min_length=2)
+    recipientOrganization: RecipientOrganizationOin
     recipientScope: str = Field(..., min_length=2)
 
     @field_validator("encryptedPersonalId")
@@ -70,7 +61,7 @@ def validate_base64(cls, v: str) -> str:
 
 class RidExchangeRequest(BaseModel):
     personalId: Any
-    recipientOrganization: str
+    recipientOrganization: RecipientOrganizationOin
     recipientScope: str
     ridUsage: Any
 
@@ -99,7 +90,7 @@ class ExchangeRequest(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
     personalId: Any
-    recipientOrganization: str
+    recipientOrganization: RecipientOrganizationOin
     recipientScope: str = Field(..., min_length=1, max_length=100)
     pseudonymType: PseudonymType