Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,29 @@
and deprecating a mistaken duplicate restores the previous Approved
entry. The `language_model_id DESC` tiebreak makes equal-created_at
rows deterministic.

## Enum coercion

`status` is stored as a `TEXT` column and typed as a `Literal` alias
on the port's `LanguageModelLookupResult` (to keep
`cora.infrastructure.ports.language_model_lookup` import-free of
Agent BC types). The adapter constructs
`LanguageModelStatus(row["status"])` as a validation step: a
corrupted row whose `status` is not a known enum value surfaces as
`ValueError` from the adapter rather than as a silent wrong-status
match downstream. `.value` on the validated member narrows to
exactly the port's alias, so no cast is needed. The SQL filter
narrows to `'Approved'`; the alias still pins the full five-value
enum because a query filter is adapter behavior, not a constraint on
the field's type.
"""

from __future__ import annotations

# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
from typing import TYPE_CHECKING

from cora.agent.aggregates.language_model import LanguageModelStatus
from cora.infrastructure.ports.language_model_lookup import LanguageModelLookupResult

if TYPE_CHECKING:
Expand Down Expand Up @@ -47,7 +63,7 @@ async def find_by_model(
return None
return LanguageModelLookupResult(
language_model_id=row["language_model_id"],
status=row["status"],
status=LanguageModelStatus(row["status"]).value,
data_tier=row["data_tier"],
archivability=row["archivability"],
snapshot_pin=row["snapshot_pin"],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,19 @@
not a single canonical row. Lives in `cora.data.adapters` because it reads a
Data-owned projection; it implements the infrastructure port so the Run BC
consumes it without importing anything Data-internal.

## Enum coercion

`status` is stored as a `TEXT` column and typed as a `Literal` alias on the
port's `DatasetDistributionLookupResult` (to keep
`cora.infrastructure.ports.dataset_distribution_lookup` import-free of Data
BC types). The adapter constructs `DistributionStatus(row["status"])` as a
validation step: a corrupted row whose `status` is not a known enum value
surfaces as `ValueError` from the adapter rather than as a silent
wrong-status match downstream. `.value` on the validated member narrows to
exactly the port's alias, so no cast is needed. The SQL filter excludes
Discarded rows; the alias still pins the full four-value enum because a
query filter is adapter behavior, not a constraint on the field's type.
"""

# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
Expand All @@ -22,6 +35,7 @@

import asyncpg

from cora.data.aggregates.distribution import DistributionStatus
from cora.infrastructure.ports.dataset_distribution_lookup import (
DatasetDistributionLookupResult,
)
Expand Down Expand Up @@ -55,7 +69,7 @@ async def find_by_datasets(
distribution_id=row["distribution_id"],
dataset_id=row["dataset_id"],
supply_id=row["supply_id"],
status=row["status"],
status=DistributionStatus(row["status"]).value,
)
)
return {dataset_id: tuple(results) for dataset_id, results in grouped.items()}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -38,8 +38,11 @@
`EnclosureLifecycle(row["lifecycle"])` as a validation step: a
corrupted row whose value is not a known enum surfaces as
`ValueError` from the adapter rather than as a silent wrong-status
match downstream. Both validated `StrEnum` values are `IS-A str`,
so assignment into the dataclass's `str`-typed fields is exact.
match downstream. The port's fields are `Literal` aliases rather
than bare `str`, and `.value` on a validated member narrows to
exactly that alias, so no cast is needed: the constructor rejects
an unknown value at runtime and the type checker confirms the
remainder statically.

## Timestamp coercion

Expand Down Expand Up @@ -140,8 +143,8 @@ def _row_to_reference(row: Any) -> EnclosureLookupResult:
return EnclosureLookupResult(
enclosure_id=row["enclosure_id"],
name=str(row["name"]),
permit_status=EnclosurePermitStatus(row["permit_status"]),
lifecycle=EnclosureLifecycle(row["lifecycle"]),
permit_status=EnclosurePermitStatus(row["permit_status"]).value,
lifecycle=EnclosureLifecycle(row["lifecycle"]).value,
permit_status_changed_at=_format_changed_at(row["last_permit_status_changed_at"]),
source_kind=row["last_source_kind"],
source_id=row["last_source_id"],
Expand Down
14 changes: 13 additions & 1 deletion apps/api/src/cora/equipment/adapters/postgres_assembly_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,17 @@
handler so the role_kind satisfaction check ORs-in the Assembly
path on top of the Family disjunction (see [[project-role-aggregate-design]]
sub-slice 3C/3D for the worked Microscope-Assembly example).

## Enum coercion

`status` is stored as a `TEXT` column and typed as a `Literal` alias
on the port's `AssemblyLookupResult` (to keep
`cora.infrastructure.ports.assembly_lookup` import-free of Equipment
BC types). The adapter constructs `AssemblyStatus(row["status"])` as
a validation step: a corrupted row whose `status` is not a known
enum value surfaces as `ValueError` from the adapter rather than as
a silent wrong-status match downstream. `.value` on the validated
member narrows to exactly the port's alias, so no cast is needed.
"""

# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
Expand All @@ -14,6 +25,7 @@

import asyncpg

from cora.equipment.aggregates.assembly import AssemblyStatus
from cora.infrastructure.ports.assembly_lookup import AssemblyLookupResult

_LOOKUP_SQL = """
Expand Down Expand Up @@ -42,7 +54,7 @@ def _row_to_result(row: Any) -> AssemblyLookupResult:
return AssemblyLookupResult(
id=row["assembly_id"],
name=str(row["name"]),
status=str(row["status"]),
status=AssemblyStatus(row["status"]).value,
presents_as=frozenset(row["presents_as"] or ()),
)

Expand Down
11 changes: 5 additions & 6 deletions apps/api/src/cora/equipment/adapters/postgres_asset_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,15 +37,14 @@
## Enum coercion

`tier` and `lifecycle` are stored as `TEXT` columns and typed as
`str` on the port's `AssetLookupResult` (to keep
`Literal` aliases on the port's `AssetLookupResult` (to keep
`cora.infrastructure.ports.asset_lookup` import-free of Equipment
BC types). The adapter still constructs `AssetTier(row["tier"])`
/ `AssetLifecycle(row["lifecycle"])` as a validation step: a
corrupted row whose `tier` or `lifecycle` is not a known enum
value surfaces as `ValueError` from the adapter rather than as a
silent wrong-tier match downstream. The validated `StrEnum` value
IS-A `str`, so the assignment into the dataclass's `str`-typed
fields is exact.
silent wrong-tier match downstream. `.value` on the validated
member narrows to exactly the port's alias, so no cast is needed.
"""

# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
Expand Down Expand Up @@ -194,8 +193,8 @@ def _row_to_result(row: Any) -> AssetLookupResult:
return AssetLookupResult(
id=row["asset_id"],
name=str(row["name"]),
tier=AssetTier(row["tier"]),
lifecycle=AssetLifecycle(row["lifecycle"]),
tier=AssetTier(row["tier"]).value,
lifecycle=AssetLifecycle(row["lifecycle"]).value,
family_affordances=frozenset(str(a) for a in row["family_affordances"]),
located_in_enclosure_id=row["located_in_enclosure_id"],
)
Expand Down
14 changes: 13 additions & 1 deletion apps/api/src/cora/equipment/adapters/postgres_family_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,17 @@
`Kernel.family_lookup` port. The 3B slice ships this adapter; 3D
wires its `bind_plan_role` handler against it for the role_kind
satisfaction-check path (Lock 17 ANY-single-family disjunction).

## Enum coercion

`status` is stored as a `TEXT` column and typed as a `Literal` alias
on the port's `FamilyLookupResult` (to keep
`cora.infrastructure.ports.family_lookup` import-free of Equipment
BC types). The adapter constructs `FamilyStatus(row["status"])` as a
validation step: a corrupted row whose `status` is not a known enum
value surfaces as `ValueError` from the adapter rather than as a
silent wrong-status match downstream. `.value` on the validated
member narrows to exactly the port's alias, so no cast is needed.
"""

# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
Expand All @@ -13,6 +24,7 @@

import asyncpg

from cora.equipment.aggregates.family import FamilyStatus
from cora.infrastructure.ports.family_lookup import FamilyLookupResult

_LOOKUP_SQL = """
Expand Down Expand Up @@ -41,7 +53,7 @@ def _row_to_result(row: Any) -> FamilyLookupResult:
return FamilyLookupResult(
id=row["family_id"],
name=str(row["name"]),
status=str(row["status"]),
status=FamilyStatus(row["status"]).value,
affordances=frozenset(row["affordances"] or ()),
presents_as=frozenset(row["presents_as"] or ()),
)
Expand Down
10 changes: 6 additions & 4 deletions apps/api/src/cora/federation/adapters/in_memory_permit_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,10 @@
from uuid import UUID

from cora.infrastructure.ports.federation.permit_lookup import (
AbiTierValue,
PermitLookup,
PermitLookupResult,
PermitStatusValue,
)
from cora.shared.facility_code import FacilityCode

Expand Down Expand Up @@ -62,8 +64,8 @@ def register_outbound(
peer_facility_id: str | FacilityCode,
artifact_kind: str,
permit_id: UUID,
status: str = "Active",
abi_tier_floor: str = "Stable",
status: PermitStatusValue = "Active",
abi_tier_floor: AbiTierValue = "Stable",
current_version: int = 0,
) -> PermitLookupResult:
"""Convenience: seed an outbound permit; returns the seeded result for assertions."""
Expand All @@ -90,8 +92,8 @@ def register_inbound(
peer_facility_id: str | FacilityCode,
artifact_kind: str,
permit_id: UUID,
status: str = "Active",
abi_tier_floor: str = "Stable",
status: PermitStatusValue = "Active",
abi_tier_floor: AbiTierValue = "Stable",
current_version: int = 0,
) -> PermitLookupResult:
"""Convenience: seed an inbound permit; returns the seeded result for assertions."""
Expand Down
23 changes: 13 additions & 10 deletions apps/api/src/cora/federation/adapters/postgres_credential_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,16 +26,19 @@

## Enum coercion

`purpose` and `status` are stored as `TEXT` columns and are
typed as `str` on the port's `CredentialLookupResult` (to keep
`purpose` and `status` are stored as `TEXT` columns. `purpose` stays
typed `str` on the port's `CredentialLookupResult` (to keep
`cora.infrastructure.ports.credential_lookup` import-free of
Federation BC types). The adapter still constructs
`CredentialPurpose(row["purpose"])` / `CredentialStatus(row["status"])`
as a validation step: a corrupted row whose `purpose` or `status`
is not a known enum value surfaces as `ValueError` from the
adapter rather than as a silent wrong-purpose match downstream.
The validated `StrEnum` value is `IS-A str`, so the assignment
into the dataclass's `str`-typed fields is exact.
Federation BC types); the adapter constructs
`CredentialPurpose(row["purpose"])` as a validation step, and the
validated `StrEnum` value IS-A `str`, so the assignment into the
`str`-typed field is exact. `status` is typed as a `Literal` alias
pinning `CredentialStatus`'s value set; the adapter constructs
`CredentialStatus(row["status"])` the same way, but takes `.value`
so the result narrows to exactly the alias, no cast needed. Either
way, a corrupted row whose `purpose` or `status` is not a known
enum value surfaces as `ValueError` from the adapter rather than as
a silent wrong-value match downstream.
"""

# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
Expand Down Expand Up @@ -79,7 +82,7 @@ def _row_to_result(row: Any) -> CredentialLookupResult:
id=row["credential_id"],
facility_id=FacilityCode(str(row["facility_id"])),
purpose=CredentialPurpose(row["purpose"]),
status=CredentialStatus(row["status"]),
status=CredentialStatus(row["status"]).value,
)


Expand Down
21 changes: 12 additions & 9 deletions apps/api/src/cora/federation/adapters/postgres_facility_lookup.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,15 +33,18 @@

## Enum coercion

`kind` and `status` are stored as `TEXT` columns and are typed as
`kind` and `status` are stored as `TEXT` columns. `kind` stays typed
`str` on the port's `FacilityLookupResult` (to keep
`cora.infrastructure.ports.facility_lookup` import-free of Federation
BC types). The adapter still constructs `FacilityKind(row["kind"])` /
`FacilityStatus(row["status"])` as a validation step: a corrupted
row whose `kind` or `status` is not a known enum value surfaces as
`ValueError` from the adapter rather than as a silent wrong-kind
match downstream. The validated `StrEnum` value is `IS-A str`, so
the assignment into the dataclass's `str`-typed fields is exact.
BC types); the adapter constructs `FacilityKind(row["kind"])` as a
validation step, and the validated `StrEnum` value IS-A `str`, so
the assignment into the `str`-typed field is exact. `status` is
typed as a `Literal` alias pinning `FacilityStatus`'s value set; the
adapter constructs `FacilityStatus(row["status"])` the same way, but
takes `.value` so the result narrows to exactly the alias, no cast
needed. Either way, a corrupted row whose `kind` or `status` is not
a known enum value surfaces as `ValueError` from the adapter rather
than as a silent wrong-value match downstream.

## JSONB array decoding

Expand Down Expand Up @@ -129,8 +132,8 @@ def _row_to_result(row: Any) -> FacilityLookupResult:
return FacilityLookupResult(
id=row["facility_id"],
code=FacilityCode(str(row["code"])),
kind=FacilityKind(row["kind"]),
status=FacilityStatus(row["status"]),
kind=FacilityKind(row["kind"]).value,
status=FacilityStatus(row["status"]).value,
trust_anchor_credential_ids=_decode_trust_anchor_ids(row["trust_anchor_credential_ids"]),
)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from threading import Lock
from uuid import UUID

from cora.infrastructure.ports.assembly_lookup import AssemblyLookupResult
from cora.infrastructure.ports.assembly_lookup import AssemblyLookupResult, AssemblyStatusValue


class InMemoryAssemblyLookup:
Expand All @@ -30,7 +30,7 @@ def register(
self,
assembly_id: UUID,
name: str,
status: str = "Defined",
status: AssemblyStatusValue = "Defined",
presents_as: Iterable[UUID] = (),
) -> None:
"""Test helper: install an Assembly summary keyed by `assembly_id`."""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@
from cora.infrastructure.ports.asset_lookup import (
ANCESTOR_WALK_DEPTH_CAP,
AncestorWalkDepthExceededError,
AssetLifecycleValue,
AssetLookupResult,
AssetTierValue,
)


Expand All @@ -37,8 +39,8 @@ def register(
self,
asset_id: UUID,
name: str,
tier: str = "Unit",
lifecycle: str = "Active",
tier: AssetTierValue = "Unit",
lifecycle: AssetLifecycleValue = "Active",
family_affordances: frozenset[str] | None = None,
parent_id: UUID | None = None,
located_in_enclosure_id: UUID | None = None,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@
from threading import Lock
from uuid import UUID

from cora.infrastructure.ports.clearance_template_lookup import ClearanceTemplateLookupResult
from cora.infrastructure.ports.clearance_template_lookup import (
ClearanceTemplateLookupResult,
ClearanceTemplateStatusValue,
)


class InMemoryClearanceTemplateLookup:
Expand All @@ -34,7 +37,7 @@ def register(
*,
facility_code: str = "aps",
code: str = "default-template",
status: str = "Active",
status: ClearanceTemplateStatusValue = "Active",
version: int = 1,
) -> None:
"""Test helper: install a clearance-template summary keyed by `template_id`.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@
from threading import Lock
from uuid import UUID

from cora.infrastructure.ports.credential_lookup import CredentialLookupResult
from cora.infrastructure.ports.credential_lookup import (
CredentialLookupResult,
CredentialStatusValue,
)
from cora.shared.facility_code import FacilityCode


Expand All @@ -32,7 +35,7 @@ def register(
credential_id: UUID,
facility_id: str | FacilityCode,
purpose: str,
status: str,
status: CredentialStatusValue,
) -> None:
"""Test helper: install a credential summary keyed by `credential_id`.

Expand Down
Loading
Loading