Python API¶

class OGCRecord(RecordMetadataMixin, pystac.Item):
    """Item-compatible Record, with OGC serialization as the default.

    Initializes STACObject directly to avoid Item's mandatory temporal extent.
    This compatibility seam must be reviewed for future PySTAC releases.
    `time` and STAC datetime fields are independent: no implicit conversion.
    """

    SCHEMA_URI = (
        "https://schemas.opengis.net/ogcapi/records/part1/1.0/openapi/schemas/recordGeoJSON.yaml"
    )
    _RESERVED = frozenset(
        {
            "id",
            "type",
            "geometry",
            "properties",
            "links",
            "assets",
            "bbox",
            "collection",
            "stac_extensions",
        }
    )

    # Preserve the positional PySTAC Item constructor API and OGC keyword options.
    def __init__(  # noqa: PLR0917
        self,
        id: str | int,
        geometry: dict[str, Any] | None = None,
        bbox: list[float] | None = None,
        datetime: Datetime | None = None,
        properties: dict[str, Any] | None = None,
        start_datetime: Datetime | None = None,
        end_datetime: Datetime | None = None,
        stac_extensions: list[str] | None = None,
        href: str | None = None,
        collection: str | pystac.Collection | None = None,
        extra_fields: dict[str, Any] | None = None,
        assets: dict[str, pystac.Asset] | None = None,
        *,
        time: dict[str, Any] | None = None,
        conforms_to: list[str] | None = None,
        link_templates: list[dict[str, Any]] | None = None,
    ) -> None:
        """Initialize a record, copying metadata and assigning asset ownership.

        OGC ``time`` is independent of the STAC datetime arguments. Optional
        OGC fields override matching entries in ``extra_fields`` when supplied.
        STAC timestamps from arguments or properties are normalized to UTC whole
        seconds; naive datetimes are treated as UTC and fractions are truncated.

        Raises:
            TypeError: If the identifier or a STAC timestamp has an invalid type.
            ValueError: If extra fields override managed record fields or a STAC
                timestamp string is invalid.
        """
        if isinstance(id, bool) or not isinstance(id, (str, int)):
            raise TypeError("Record id must be a string or integer")
        pystac.STACObject.__init__(self, list(stac_extensions or []))
        self._record_id = id
        self.id = str(id)  # PySTAC path/layout utilities require a string.
        self.geometry = deepcopy(geometry)
        self.bbox = deepcopy(bbox)
        self._null_properties = properties is None
        self.properties = deepcopy(properties) if properties is not None else {}
        self.extra_fields = deepcopy(extra_fields) if extra_fields else {}
        if self._RESERVED.intersection(self.extra_fields):
            raise ValueError("extra_fields must not override managed fields")
        self.assets = {}
        self.collection_id = None
        self._stac_io = None
        self._set_temporal_properties(datetime, start_datetime, end_datetime)
        for field_name, field_value in (
            ("time", time),
            ("conformsTo", conforms_to),
            ("linkTemplates", link_templates),
        ):
            if field_value is not None:
                self.extra_fields[field_name] = deepcopy(field_value)
        self._attach_stac_objects(collection, assets, href)

    def _attach_stac_objects(
        self,
        collection: str | pystac.Collection | None,
        assets: dict[str, pystac.Asset] | None,
        href: str | None,
    ) -> None:
        """Set collection links, asset ownership, and the record's location."""
        if isinstance(collection, pystac.Collection):
            self.set_collection(collection)
        elif collection is not None:
            self.collection_id = collection
        for key, asset in (assets or {}).items():
            self.add_asset(key, asset)
        if href is not None:
            self.set_self_href(href)

    def _set_temporal_properties(
        self,
        instant: Datetime | None,
        start: Datetime | None,
        end: Datetime | None,
    ) -> None:
        """Apply explicit STAC dates and normalize property fallbacks to UTC seconds.

        Non-null constructor arguments override the corresponding properties.
        """
        for field_name, value in (
            ("datetime", instant),
            ("start_datetime", start),
            ("end_datetime", end),
        ):
            if value is not None:
                self.properties[field_name] = value
        _normalize_temporal_properties(self.properties)
        stored_instant = self.properties.get("datetime")
        self.datetime = None if stored_instant is None else _metadata_datetime(stored_instant)

    @property
    def record_id(self) -> str | int:
        return self._record_id if str(self._record_id) == self.id else self.id

    @property
    def record_metadata(self) -> RecordCommonProperties:
        return RecordCommonProperties(self)

    @property
    def time(self) -> dict[str, Any] | None:
        return self.extra_fields.get("time")

    @time.setter
    def time(self, value: dict[str, Any] | None) -> None:
        self.extra_fields["time"] = value

    @property
    def conforms_to(self) -> list[str]:
        """The live list of conformance URIs.

        Raises:
            TypeError: If the stored value is not a list of strings.
        """
        conformance = self.extra_fields.setdefault("conformsTo", [])
        if not isinstance(conformance, list) or not all(
            isinstance(uri, str) for uri in conformance
        ):
            raise TypeError("conformsTo must be a list of strings")
        return conformance

    @conforms_to.setter
    def conforms_to(self, value: list[str]) -> None:
        self.extra_fields["conformsTo"] = value

    @property
    def link_templates(self) -> list[dict[str, Any]]:
        """The live list of link template dictionaries.

        Raises:
            TypeError: If the stored value is not a list of dictionaries.
        """
        templates = self.extra_fields.setdefault("linkTemplates", [])
        if not isinstance(templates, list) or not all(
            isinstance(template, dict) for template in templates
        ):
            raise TypeError("linkTemplates must be a list of dictionaries")
        return templates

    @link_templates.setter
    def link_templates(self, value: list[dict[str, Any]]) -> None:
        self.extra_fields["linkTemplates"] = value

    def to_dict(
        self, include_self_link: bool = True, transform_hrefs: bool = True
    ) -> dict[str, Any]:
        """Serialize a record with metadata and STAC timestamps in UTC whole seconds.

        Normalization includes nested metadata and does not mutate the record.

        Raises:
            ValueError: If a metadata or STAC timestamp string is invalid.
            TypeError: If a non-null timestamp is neither a datetime nor a string.
        """
        props = deepcopy(self.properties)
        if self.datetime is not None:
            props["datetime"] = self.datetime
        _normalize_temporal_properties(props)
        doc = deepcopy(self.extra_fields)
        doc.update(
            type="Feature",
            id=self.record_id,
            geometry=deepcopy(self.geometry),
            properties=None if self._null_properties and not props else props,
            links=[
                deepcopy(link.to_dict(transform_href=transform_hrefs))
                for link in self.links
                if include_self_link or link.rel != pystac.RelType.SELF
            ],
        )
        if self.bbox is not None:
            doc["bbox"] = list(self.bbox)
        # Preserve extension declarations/assets as GeoJSON foreign members.
        # Never substitute conformsTo for stac_extensions.
        if self.stac_extensions:
            doc["stac_extensions"] = list(self.stac_extensions)
        if self.assets:
            doc["assets"] = {key: deepcopy(asset.to_dict()) for key, asset in self.assets.items()}
        if self.collection_id is not None:
            doc["collection"] = self.collection_id
        _normalize_metadata_timestamps(doc)
        return doc

    def to_record_dict(
        self, include_self_link: bool = True, transform_hrefs: bool = True
    ) -> dict[str, Any]:
        return self.to_dict(include_self_link, transform_hrefs)

    @classmethod
    def matches_object_type(cls, d: dict[str, Any]) -> bool:
        # Structural recognition only; GeoJSON/STAC overlap prevents unique dispatch.
        return (
            d.get("type") == "Feature"
            and all(key in d for key in ("id", "geometry", "properties"))
            and isinstance(d["id"], (str, int))
            and not isinstance(d["id"], bool)
            and (d["properties"] is None or isinstance(d["properties"], dict))
        )

    @classmethod
    def from_dict(
        cls,
        d: dict[str, Any],
        href: str | None = None,
        root: pystac.Catalog | None = None,
        migrate: bool = True,
        preserve_dict: bool = True,
    ) -> OGCRecord:
        """Read Records; migrate is accepted for compatibility but never applied.

        Always copies document metadata, even when preserve_dict=False.
        """
        if not cls.matches_object_type(d):
            raise ValueError("Expected a Record GeoJSON Feature with id, geometry, properties")
        obj = cls(
            id=d["id"],
            geometry=d["geometry"],
            properties=d["properties"],
            bbox=d.get("bbox"),
            collection=d.get("collection"),
            stac_extensions=d.get("stac_extensions"),
            extra_fields={k: v for k, v in d.items() if k not in cls._RESERVED},
            assets={k: pystac.Asset.from_dict(deepcopy(v)) for k, v in d.get("assets", {}).items()},
        )
        for link in d.get("links", []):
            if href is None or link.get("rel") != "self":
                obj.add_link(pystac.Link.from_dict(deepcopy(link)))
        if href is not None:
            obj.set_self_href(href)
        if root is not None:
            obj.set_root(root)
        return obj

    def clone(self) -> OGCRecord:
        doc = self.to_dict(transform_hrefs=False)
        doc["links"] = []
        result = type(self).from_dict(doc)
        for link in self.links:
            cloned_link = link.clone()
            cloned_link.extra_fields = deepcopy(link.extra_fields)
            result.add_link(cloned_link)
        result._stac_io = self._stac_io
        return result

    def to_stac_item(self) -> pystac.Item:
        """Explicit export; requires a genuine STAC temporal extent.

        Metadata and STAC timestamps are normalized to UTC whole seconds.
        Constructing an Item does not replace full STAC schema validation.
        """
        if self.datetime is None and not all(
            self.properties.get(k) is not None for k in ("start_datetime", "end_datetime")
        ):
            raise ValueError("STAC export requires datetime or start_datetime/end_datetime")
        extra = deepcopy(self.extra_fields)
        extra.pop("stac_version", None)
        item = pystac.Item(
            id=self.id,
            geometry=deepcopy(self.geometry),
            bbox=deepcopy(self.bbox),
            datetime=None if self.datetime is None else _metadata_datetime(self.datetime),
            properties=deepcopy(self.properties),
            stac_extensions=list(self.stac_extensions),
            collection=self.collection_id,
            extra_fields=extra,
            assets={k: v.clone() for k, v in self.assets.items()},
        )
        for link in self.links:
            cloned_link = link.clone()
            cloned_link.extra_fields = deepcopy(link.extra_fields)
            item.add_link(cloned_link)
        _normalize_temporal_properties(item.properties)
        _normalize_metadata_timestamps(item.properties)
        _normalize_metadata_timestamps(item.extra_fields)
        for asset in item.assets.values():
            _normalize_metadata_timestamps(asset.extra_fields)
        for link in item.links:
            _normalize_metadata_timestamps(link.extra_fields)
        return item

    def validate(self, validator: Any = None) -> list[Any]:
        """Validate the OGC document with an explicitly supplied validator.

        Supply an object exposing validate(document), configured for the OGC
        OpenAPI 3.0 schema and its references. No implicit STAC validation.
        """
        if validator is None:
            raise ValueError(
                "Supply an OGC schema validator; for STAC use record.to_stac_item().validate()"
            )
        validator.validate(self.to_dict(transform_hrefs=False))
        return [self.SCHEMA_URI]

    def __repr__(self) -> str:
        return f"<OGCRecord id={self.record_id!r}>"

class RecordCommonProperties(RecordMetadataMixin):
    """Backward-compatible metadata view sharing the record properties."""

    def __init__(self, record: OGCRecord) -> None:
        self.record = record

    @property
    def properties(self) -> dict[str, Any]:
        """The live metadata dictionary shared with the record."""
        return self.record.properties

    @properties.setter
    def properties(self, value: dict[str, Any]) -> None:
        self.record.properties = value

    def to_dict(self) -> dict[str, Any]:
        """Copy properties with metadata and STAC timestamps normalized for JSON."""
        properties = deepcopy(self.properties)
        _normalize_temporal_properties(properties)
        _normalize_metadata_timestamps(properties)
        return properties

class RecordMetadataMixin:
    """Record metadata accessors with datetime values for creation and update times.

    Nested structures remain open dictionaries, allowing future extensions.
    These accessors do not perform JSON Schema validation.
    """

    # PySTAC properties are an open JSON metadata boundary.
    properties: dict[str, Any]

    @property
    def created(self) -> Datetime | None:
        """The created timestamp in UTC whole seconds, or None if absent.

        Assign a datetime to store its ISO 8601 string, or None to remove it.
        Naive datetimes are treated as UTC; fractional seconds are truncated.

        Raises:
            ValueError: If the stored timestamp string is invalid.
            TypeError: If the stored value is neither a datetime nor a string.
        """
        value = self.properties.get("created")
        return None if value is None else _metadata_datetime(value)

    @created.setter
    def created(self, value: Datetime | None) -> None:
        if value is None:
            self.properties.pop("created", None)
        else:
            self.properties["created"] = datetime_to_str(
                _metadata_datetime(value), timespec="seconds"
            )

    @property
    def updated(self) -> Datetime | None:
        """The updated timestamp in UTC whole seconds, or None if absent.

        Assign a datetime to store its ISO 8601 string, or None to remove it.
        Naive datetimes are treated as UTC; fractional seconds are truncated.

        Raises:
            ValueError: If the stored timestamp string is invalid.
            TypeError: If the stored value is neither a datetime nor a string.
        """
        value = self.properties.get("updated")
        return None if value is None else _metadata_datetime(value)

    @updated.setter
    def updated(self, value: Datetime | None) -> None:
        if value is None:
            self.properties.pop("updated", None)
        else:
            self.properties["updated"] = datetime_to_str(
                _metadata_datetime(value), timespec="seconds"
            )

    type = MetadataField[str]("type")
    title = MetadataField[str]("title")
    description = MetadataField[str]("description")
    keywords = MetadataField[list[str]]("keywords")
    themes = MetadataField[list[Theme]]("themes")
    language = MetadataField[Language]("language")
    languages = MetadataField[list[Language]]("languages")
    resource_languages = MetadataField[list[Language]]("resourceLanguages")
    external_ids = MetadataField[list[ExternalId]]("externalIds")
    formats = MetadataField[list[Format]]("formats")
    contacts = MetadataField[list[Contact]]("contacts")
    license = MetadataField[str]("license")
    rights = MetadataField[str]("rights")