ENH: Add an endpoint to fetch well headers from SMDA

src/fmu_settings_api/interfaces/smda_api.py

@@ -21,6 +21,7 @@ class SmdaRoutes:
     STRAT_UNITS_SEARCH: Final[str] = "smda-api/strat-units/search"
     COORDINATE_SYSTEM_SEARCH: Final[str] = "smda-api/crs/search"
     SURFACE_SEARCH: Final[str] = "smda-api/strat-surface-names/search"
+    WELL_HEADER_SEARCH: Final[str] = "smda-api/wellheader-insight-headers/search"
 
 
 class SmdaAPI:
@@ -113,6 +114,16 @@ async def discovery(
             json={"_projection": _projection, "field_identifier": field_identifiers},
         )
 
+    async def well_headers(
+        self, field_identifiers: Sequence[str], columns: Sequence[str] | None = None
+    ) -> httpx2.Response:
+        """Searches for well headers related to a field identifier."""
+        _projection = "identifier,uuid" if columns is None else ",".join(columns)
+        return await self.post(
+            SmdaRoutes.WELL_HEADER_SEARCH,
+            json={"_projection": _projection, "field_identifier": field_identifiers},
+        )
+
     async def strat_column_areas(
         self, field_identifiers: Sequence[str], columns: Sequence[str] | None = None
     ) -> httpx2.Response:

src/fmu_settings_api/models/smda.py

@@ -63,6 +63,101 @@ class SmdaFieldSearchResult(BaseResponseModel):
     """A list of field identifier results from the search."""
 
 
+class SmdaWellHeader(BaseResponseModel):
+    """Well header data from SMDA."""
+
+    unique_well_identifier: str
+    """Unique SMDA identifier for the well."""
+
+    unique_wellbore_identifier: str
+    """Unique SMDA identifier for the wellbore."""
+
+    official_wellbore_name: str | None
+    """Official wellbore name used by the Authorities.
+
+    For Norway and UK, it will be the unique_wellbore_identifier without
+    country iso code, but for Brazil it can really differs from the Equinor
+    wellbore name.
+    """
+
+    country_identifier: str
+    """Country identifier for the wellbore."""
+
+    parent_wellbore: str | None
+    """The unique wellbore identifier this wellbore is kicked off from.
+
+    Ref. kick off depth. This is used for sidetracks. A wellbore starting at
+    the well origin has no parent.
+    """
+
+    wellbore_type: str | None
+    """Type of wellbore, values like exploration, development, other.
+
+    This attribute is automatically maintained in SMDA based on the wellbore
+    purpose. If the purpose is like wildcat or appraisal, type will be set to
+    exploration, if the purpose is like production, injection then the type is
+    set to development.
+    """
+
+    wellbore_purpose: str | None
+    """Purpose of wellbore.
+
+    Values like wildcat, appraisal, … for exploration wellbores; production,
+    injection, observation, disposal, … for development wellbores; shallow gas,
+    pilot hole for other purpose.
+    """
+
+    wellbore_status: str | None
+    """Status of the wellbore.
+
+    Value like plugged and abandoned, drilling, plugged, producing ... This
+    attribute is automatically maintained in SMDA if no good source is found
+    for it. SMDA will use the wellbore type (exploration or development), the
+    drill dates information, current_track, etc ... in order to set a plausible
+    status. If wellbore type=exploration and completed_date < current_date,
+    then status=plugged and abandoned while development wellbore would be set
+    to completed.
+    """
+
+    wellbore_purpose_planned: str | None
+    """Pre-drill purpose of the wellbore.
+
+    Legal values for exploration wellbores: wildcat, appraisal. Example of
+    legal values for development wellbores: observation, production, injection.
+    """
+
+    drill_year: int | None
+    """The year when the drilling has started."""
+
+    completion_date: str | None
+    """Date when the wellbore is considered completed.
+
+    For exploration wellbores from moveable facilities, this may be the anchor
+    handling or jacking-down start date. For fixed facilities and development
+    wellbores, it is when the wellbore reaches total depth and the last casing,
+    liner, or screen is set. If immediately plugged, it is the date the last
+    plug is set.
+    """
+
+    discovery_internal_identifier: str | None
+    """Internal name of the discovery."""
+
+    multilateral: Literal[0, 1] | None
+    """Whether the wellbore is multilateral. 0 = no, 1 = yes."""
+
+    projected_coordinate_unit: str | None
+    """Projected coordinate unit."""
+
+    projected_coordinate_system: str | None
+    """Coordinate reference system for the easting/northing values."""
+
+    well_uuid: UUID
+    """SMDA UUID for the well."""
+
+    wellbore_uuid: UUID
+    """SMDA UUID for the wellbore."""
+
+
 class SmdaMasterdataResult(BaseResponseModel):
     """Contains SMDA-related attributes."""
 
@@ -152,3 +247,10 @@ class SmdaStratigraphicUnitsResult(BaseResponseModel):
 
     stratigraphic_units: list[StratigraphicUnit]
     """List of stratigraphic units from SMDA."""
+
+
+class SmdaWellHeadersResult(BaseResponseModel):
+    """Result containing a list of well headers."""
+
+    well_headers: list[SmdaWellHeader]
+    """List of well headers from SMDA."""

src/fmu_settings_api/services/smda.py

@@ -3,6 +3,7 @@
 import asyncio
 from collections.abc import Sequence
 from typing import Any, Final
+from uuid import NAMESPACE_URL, uuid5
 
 import httpx2
 from fmu.datamodels.common.masterdata import (
@@ -14,6 +15,7 @@
 )
 from fmu.settings._drogon import (
     MASTERDATA as DROGON_MASTERDATA,
+    RMS_WELLS as DROGON_RMS_WELLS,
     RMS_ZONES as DROGON_RMS_ZONES,
     STRATIGRAPHY_MAPPINGS as DROGON_STRATIGRAPHY_MAPPINGS,
 )
@@ -27,6 +29,8 @@
     SmdaMasterdataResult,
     SmdaSelectedField,
     SmdaStratigraphicUnitsResult,
+    SmdaWellHeader,
+    SmdaWellHeadersResult,
     StratigraphicUnit,
 )
 
@@ -68,6 +72,31 @@
     for zone in DROGON_RMS_ZONES
     if zone["name"] in DROGON_STRATIGRAPHY_BY_SOURCE_ID
 ]
+DROGON_WELL_HEADERS: Final[list[SmdaWellHeader]] = [
+    SmdaWellHeader(
+        unique_well_identifier=f"NO {well['name']}",
+        unique_wellbore_identifier=f"NO {well['name']}",
+        official_wellbore_name=str(well["name"]),
+        country_identifier=DROGON_SMDA_MASTERDATA["country"][0]["identifier"],
+        parent_wellbore=None,
+        wellbore_type="development",
+        wellbore_purpose="production",
+        wellbore_status="operating",
+        wellbore_purpose_planned="production",
+        drill_year=None,
+        completion_date=None,
+        discovery_internal_identifier=None,
+        multilateral=1 if well["name"] == "MLW_OP5_Y1" else 0,
+        projected_coordinate_unit="m",
+        projected_coordinate_system=DROGON_SMDA_MASTERDATA["coordinate_system"][
+            "identifier"
+        ],
+        # Synthetic but stable UUIDs for the built-in Drogon data.
+        well_uuid=uuid5(NAMESPACE_URL, f"well/{well['name']}"),
+        wellbore_uuid=uuid5(NAMESPACE_URL, f"wellbore/{well['name']}"),
+    )
+    for well in DROGON_RMS_WELLS
+]
 
 
 def _is_drogon_identifier(identifier: str) -> bool:
@@ -342,6 +371,48 @@ async def get_stratigraphic_units(
                 strat_unit_items.append(strat_unit_item)
         return SmdaStratigraphicUnitsResult(stratigraphic_units=strat_unit_items)
 
+    async def get_well_headers(self, field_identifier: str) -> SmdaWellHeadersResult:
+        """Queries well headers for a field identifier."""
+        if not field_identifier:
+            raise ValueError("A field identifier must be provided")
+
+        if _is_drogon_identifier(field_identifier):
+            return SmdaWellHeadersResult(well_headers=DROGON_WELL_HEADERS)
+
+        well_header_items = []
+        well_header_res = await self._smda.well_headers(
+            [field_identifier],
+            columns=[
+                "unique_well_identifier",
+                "unique_wellbore_identifier",
+                "official_wellbore_name",
+                "country_identifier",
+                "parent_wellbore",
+                "wellbore_type",
+                "wellbore_purpose",
+                "wellbore_status",
+                "wellbore_purpose_planned",
+                "drill_year",
+                "completion_date",
+                "discovery_internal_identifier",
+                "multilateral",
+                "projected_coordinate_unit",
+                "projected_coordinate_system",
+                "well_uuid",
+                "wellbore_uuid",
+            ],
+        )
+        well_header_results = well_header_res.json()["data"]["results"]
+        for well_header_data in well_header_results:
+            well_header_item = SmdaWellHeader(**well_header_data)
+            if well_header_item not in well_header_items:
+                well_header_items.append(well_header_item)
+        if not well_header_items:
+            raise ValueError(
+                f"No well headers found for field identifier: {field_identifier}"
+            )
+        return SmdaWellHeadersResult(well_headers=well_header_items)
+
     async def _get_countries(
         self, country_identifiers: Sequence[str]
     ) -> list[CountryItem]:

src/fmu_settings_api/v1/routes/smda/main.py

@@ -20,6 +20,7 @@
     SmdaSelectedField,
     SmdaStratColumn,
     SmdaStratigraphicUnitsResult,
+    SmdaWellHeadersResult,
 )
 from fmu_settings_api.v1.responses import (
     GetSessionResponses,
@@ -34,6 +35,7 @@
         [
             {"detail": "At least one SMDA field must be provided"},
             {"detail": "A stratigraphic column identifier must be provided"},
+            {"detail": "A field identifier must be provided"},
         ],
     ),
     **inline_add_response(
@@ -55,6 +57,7 @@
         [
             {"detail": "No fields found for identifiers: {identifiers}"},
             {"detail": "No stratigraphic units found for column: {identifier}"},
+            {"detail": "No well headers found for field identifier: {identifier}"},
         ],
     ),
     **inline_add_response(
@@ -341,3 +344,61 @@ async def post_strat_units(
             detail="SMDA API request timed out. Please try again.",
             headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA},
         ) from e
+
+
+@router.post(
+    "/well_headers",
+    response_model=SmdaWellHeadersResult,
+    summary="Retrieves well headers for a field",
+    description=dedent(
+        """
+        A route to gather well header data from SMDA for a specified field.
+
+        This route receives a valid field identifier and returns the well header
+        fields needed to identify and map SMDA wellbores against model well names.
+        """
+    ),
+    responses={
+        **GetSessionResponses,
+        **SmdaUpstreamErrorResponses,
+    },
+)
+async def post_well_headers(
+    field: SmdaField,
+    smda_service: ProjectSmdaServiceDep,
+) -> SmdaWellHeadersResult:
+    """Queries SMDA well headers for a specified field."""
+    try:
+        return await smda_service.get_well_headers(field.identifier)
+    except ValueError as e:
+        error_msg = str(e)
+        match error_msg:
+            case msg if "A field identifier must be provided" in msg:
+                status_code = 400
+            case msg if "No well headers found" in msg:
+                status_code = 422
+            case _:  # pragma: no cover
+                status_code = 400
+        raise HTTPException(
+            status_code=status_code,
+            detail=error_msg,
+            headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA},
+        ) from e
+    except httpx2.HTTPStatusError as e:
+        raise HTTPException(
+            status_code=e.response.status_code,
+            detail=f"SMDA error requesting {e.request.url}",
+            headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA},
+        ) from e
+    except KeyError as e:
+        raise HTTPException(
+            status_code=500,
+            detail=f"Malformed response from SMDA: {e}",
+            headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA},
+        ) from e
+    except TimeoutError as e:
+        raise HTTPException(
+            status_code=503,
+            detail="SMDA API request timed out. Please try again.",
+            headers={HttpHeader.UPSTREAM_SOURCE_KEY: HttpHeader.UPSTREAM_SOURCE_SMDA},
+        ) from e

tests/test_interfaces/test_smda_api.py

@@ -296,3 +296,39 @@ async def test_smda_discovery_search(mock_httpx_post: MagicMock) -> None:
         },
     )
     res.raise_for_status.assert_called_once()  # type: ignore
+
+
+async def test_smda_well_header_search(mock_httpx_post: MagicMock) -> None:
+    """Tests well header search sends default payload."""
+    api = SmdaAPI("token", "key")
+
+    res = await api.well_headers(["FIELD_A"])
+
+    mock_httpx_post.assert_called_with(
+        f"{SmdaRoutes.BASE_URL}/{SmdaRoutes.WELL_HEADER_SEARCH}",
+        headers=api._headers,
+        json={
+            "_projection": "identifier,uuid",
+            "field_identifier": ["FIELD_A"],
+        },
+    )
+    res.raise_for_status.assert_called_once()  # type: ignore
+
+
+async def test_smda_well_header_search_with_columns(
+    mock_httpx_post: MagicMock,
+) -> None:
+    """Tests well header search sends requested projection."""
+    api = SmdaAPI("token", "key")
+
+    res = await api.well_headers(["FIELD_A"], columns=["unique_well_identifier"])
+
+    mock_httpx_post.assert_called_with(
+        f"{SmdaRoutes.BASE_URL}/{SmdaRoutes.WELL_HEADER_SEARCH}",
+        headers=api._headers,
+        json={
+            "_projection": "unique_well_identifier",
+            "field_identifier": ["FIELD_A"],
+        },
+    )
+    res.raise_for_status.assert_called_once()  # type: ignore