Skip to content

Update file extension and add Link.ext #1265

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Oct 17, 2023
Merged
Show file tree
Hide file tree
Changes from 2 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
8 changes: 4 additions & 4 deletions pystac/extensions/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,7 @@ def validate_owner_has_extension(
@classmethod
def ensure_owner_has_extension(
cls,
asset: pystac.Asset | AssetDefinition,
asset_or_link: pystac.Asset | AssetDefinition | pystac.Link,
add_if_missing: bool = False,
) -> None:
"""Given an :class:`~pystac.Asset`, checks if the asset's owner has this
Expand All @@ -206,15 +206,15 @@ def ensure_owner_has_extension(
STACError : If ``add_if_missing`` is ``True`` and ``asset.owner`` is
``None``.
"""
if asset.owner is None:
if asset_or_link.owner is None:
if add_if_missing:
raise pystac.STACError(
"Attempted to use add_if_missing=True for an Asset or ItemAsset "
f"Attempted to use add_if_missing=True for a {type(asset_or_link)} "
"with no owner. Use .set_owner or set add_if_missing=False."
)
else:
return
return cls.ensure_has_extension(cast(S, asset.owner), add_if_missing)
return cls.ensure_has_extension(cast(S, asset_or_link.owner), add_if_missing)

@classmethod
def validate_has_extension(cls, obj: S, add_if_missing: bool = False) -> None:
Expand Down
40 changes: 39 additions & 1 deletion pystac/extensions/ext.py
Original file line number Diff line number Diff line change
Expand Up @@ -273,7 +273,7 @@ class AssetExt(_AssetExt[pystac.Asset]):
stac_object: pystac.Asset

@property
def file(self) -> FileExtension:
def file(self) -> FileExtension[pystac.Asset]:
return FileExtension.ext(self.stac_object)

@property
Expand All @@ -288,3 +288,41 @@ def xarray(self) -> XarrayAssetsExtension[pystac.Asset]:
@dataclass
class ItemAssetExt(_AssetExt[AssetDefinition]):
stac_object: AssetDefinition


@dataclass
class LinkExt:
stac_object: pystac.Link

def has(self, name: EXTENSION_NAMES) -> bool:
if self.stac_object.owner is None:
raise pystac.STACError(
f"Attempted to add extension='{name}' for a Link with no owner. "
"Use Link.set_owner and then try to add the extension again."
)
else:
return cast(
bool, _get_class_by_name(name).has_extension(self.stac_object.owner)
)

def add(self, name: EXTENSION_NAMES) -> None:
if self.stac_object.owner is None:
raise pystac.STACError(
f"Attempted to add extension='{name}' for a Link with no owner. "
"Use Link.set_owner and then try to add the extension again."
)
else:
_get_class_by_name(name).add_to(self.stac_object.owner)

def remove(self, name: EXTENSION_NAMES) -> None:
if self.stac_object.owner is None:
raise pystac.STACError(
f"Attempted to remove extension='{name}' for a Link with no owner. "
"Use Link.set_owner and then try to remove the extension again."
)
else:
_get_class_by_name(name).remove_from(self.stac_object.owner)

@property
def file(self) -> FileExtension[pystac.Link]:
return FileExtension.ext(self.stac_object)
132 changes: 104 additions & 28 deletions pystac/extensions/file.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from __future__ import annotations

from collections.abc import Iterable
from typing import Any, Literal, Union
from typing import Any, Generic, Literal, TypeVar, Union, cast

import pystac
from pystac.extensions.base import ExtensionManagementMixin, PropertiesExtension
Expand All @@ -15,14 +15,17 @@
)
from pystac.utils import StringEnum, get_required, map_opt

SCHEMA_URI = "https://stac-extensions.github.io/file/v2.0.0/schema.json"
T = TypeVar("T", pystac.Asset, pystac.Link)

SCHEMA_URI = "https://stac-extensions.github.io/file/v2.1.0/schema.json"

PREFIX = "file:"
BYTE_ORDER_PROP = PREFIX + "byte_order"
CHECKSUM_PROP = PREFIX + "checksum"
HEADER_SIZE_PROP = PREFIX + "header_size"
SIZE_PROP = PREFIX + "size"
VALUES_PROP = PREFIX + "values"
LOCAL_PATH_PROP = PREFIX + "local_path"


class ByteOrder(StringEnum):
Expand Down Expand Up @@ -92,10 +95,13 @@ def to_dict(self) -> dict[str, Any]:


class FileExtension(
PropertiesExtension, ExtensionManagementMixin[Union[pystac.Collection, pystac.Item]]
Generic[T],
PropertiesExtension,
ExtensionManagementMixin[Union[pystac.Catalog, pystac.Collection, pystac.Item]],
):
"""A class that can be used to extend the properties of an :class:`~pystac.Asset`
with properties from the :stac-ext:`File Info Extension <file>`.
or :class:`~pystac.Link` with properties from the
:stac-ext:`File Info Extension <file>`.

To create an instance of :class:`FileExtension`, use the
:meth:`FileExtension.ext` method. For example:
Expand All @@ -108,32 +114,14 @@ class FileExtension(

name: Literal["file"] = "file"

asset_href: str
"""The ``href`` value of the :class:`~pystac.Asset` being extended."""

properties: dict[str, Any]
"""The :class:`~pystac.Asset` fields, including extension properties."""

additional_read_properties: Iterable[dict[str, Any]] | None = None
"""If present, this will be a list containing 1 dictionary representing the
properties of the owning :class:`~pystac.Item`."""

def __init__(self, asset: pystac.Asset):
self.asset_href = asset.href
self.properties = asset.extra_fields
if asset.owner and isinstance(asset.owner, pystac.Item):
self.additional_read_properties = [asset.owner.properties]

def __repr__(self) -> str:
return f"<AssetFileExtension Asset href={self.asset_href}>"

def apply(
self,
byte_order: ByteOrder | None = None,
checksum: str | None = None,
header_size: int | None = None,
size: int | None = None,
values: list[MappingObject] | None = None,
local_path: str | None = None,
) -> None:
"""Applies file extension properties to the extended Item.

Expand All @@ -154,6 +142,7 @@ def apply(
self.header_size = header_size
self.size = size
self.values = values
self.local_path = local_path

@property
def byte_order(self) -> ByteOrder | None:
Expand Down Expand Up @@ -184,6 +173,22 @@ def header_size(self) -> int | None:
def header_size(self, v: int | None) -> None:
self._set_property(HEADER_SIZE_PROP, v)

@property
def local_path(self) -> str | None:
"""Get or sets a relative local path for the asset/link.

The ``file:local_path`` field indicates a **relative** path that
can be used by clients for different purposes to organize the
files locally. For compatibility reasons the name-separator
character in paths **must** be `/` and the Windows separator `\\`
is **not** allowed.
"""
return self._get_property(LOCAL_PATH_PROP, str)

@local_path.setter
def local_path(self, v: str | None) -> None:
self._set_property(LOCAL_PATH_PROP, v, pop_if_none=True)

@property
def size(self) -> int | None:
"""Get or sets the size of the file, in bytes."""
Expand Down Expand Up @@ -220,23 +225,94 @@ def get_schema_uri(cls) -> str:
return SCHEMA_URI

@classmethod
def ext(cls, obj: pystac.Asset, add_if_missing: bool = False) -> FileExtension:
def ext(
cls, obj: pystac.Asset | pystac.Link, add_if_missing: bool = False
) -> FileExtension[T]:
"""Extends the given STAC Object with properties from the :stac-ext:`File Info
Extension <file>`.

This extension can be applied to instances of :class:`~pystac.Asset`.
This extension can be applied to instances of :class:`~pystac.Asset` or
:class:`~pystac.Link`
"""
if isinstance(obj, pystac.Asset):
cls.ensure_owner_has_extension(obj, add_if_missing)
return cls(obj)
return cast(FileExtension[T], AssetFileExtension(obj))
elif isinstance(obj, pystac.Link):
cls.ensure_owner_has_extension(obj, add_if_missing)
return cast(FileExtension[T], LinkFileExtension(obj))
else:
raise pystac.ExtensionTypeError(cls._ext_error_message(obj))


class AssetFileExtension(FileExtension[pystac.Asset]):
"""A concrete implementation of :class:`FileExtension` on an
:class:`~pystac.Asset` that extends the Asset fields to include properties defined
in the :stac-ext:`File Info Extension <file>`.

This class should generally not be instantiated directly. Instead, call
:meth:`FileExtension.ext` on an :class:`~pystac.Asset` to extend it.
"""

asset_href: str
"""The ``href`` value of the :class:`~pystac.Asset` being extended."""

properties: dict[str, Any]
"""The :class:`~pystac.Asset` fields, including extension properties."""

additional_read_properties: Iterable[dict[str, Any]] | None = None
"""If present, this will be a list containing 1 dictionary representing the
properties of the owner."""

def __init__(self, asset: pystac.Asset):
self.asset_href = asset.href
self.properties = asset.extra_fields
if asset.owner and hasattr(asset.owner, "properties"):
self.additional_read_properties = [asset.owner.properties]

def __repr__(self) -> str:
return f"<AssetFileExtension Asset href={self.asset_href}>"


class LinkFileExtension(FileExtension[pystac.Link]):
"""A concrete implementation of :class:`FileExtension` on an
:class:`~pystac.Link` that extends the Link fields to include properties defined
in the :stac-ext:`File Info Extension <file>`.

This class should generally not be instantiated directly. Instead, call
:meth:`FileExtension.ext` on an :class:`~pystac.Link` to extend it.
"""

link_href: str
"""The ``href`` value of the :class:`~pystac.Link` being extended."""

properties: dict[str, Any]
"""The :class:`~pystac.Link` fields, including extension properties."""

additional_read_properties: Iterable[dict[str, Any]] | None = None
"""If present, this will be a list containing 1 dictionary representing the
properties of the owner."""

def __init__(self, link: pystac.Link):
self.link_href = link.href
self.properties = link.extra_fields
if link.owner and hasattr(link.owner, "properties"):
self.additional_read_properties = [link.owner.properties]

def __repr__(self) -> str:
return f"<LinkFileExtension Link href={self.link_href}>"


class FileExtensionHooks(ExtensionHooks):
schema_uri: str = SCHEMA_URI
prev_extension_ids = {"file"}
stac_object_types = {pystac.STACObjectType.ITEM}
prev_extension_ids = {
"file",
"https://stac-extensions.github.io/file/v2.0.0/schema.json",
}
stac_object_types = {
pystac.STACObjectType.ITEM,
pystac.STACObjectType.COLLECTION,
pystac.STACObjectType.CATALOG,
}

def migrate(
self, obj: dict[str, Any], version: STACVersionID, info: STACJSONDescription
Expand Down
13 changes: 13 additions & 0 deletions pystac/link.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
if TYPE_CHECKING:
from pystac.catalog import Catalog
from pystac.collection import Collection
from pystac.extensions.ext import LinkExt
from pystac.item import Item
from pystac.stac_object import STACObject

Expand Down Expand Up @@ -500,3 +501,15 @@ def canonical(
title=title,
media_type=pystac.MediaType.JSON,
)

@property
def ext(self) -> LinkExt:
"""Accessor for extension classes on this link

Example::

link.ext.file.size = 8675309
"""
from pystac.extensions.ext import LinkExt

return LinkExt(stac_object=self)
8 changes: 7 additions & 1 deletion tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

import pytest

from pystac import Asset, Catalog, Collection, Item
from pystac import Asset, Catalog, Collection, Item, Link

from .utils import ARBITRARY_BBOX, ARBITRARY_EXTENT, ARBITRARY_GEOM, TestCases

Expand Down Expand Up @@ -35,6 +35,12 @@ def asset(item: Item) -> Asset:
return item.assets["foo"]


@pytest.fixture
def link(item: Item) -> Link:
item.add_link(Link(rel="child", target="https://example.tif"))
return item.links[0]


@pytest.fixture
def test_case_1_catalog() -> Catalog:
return TestCases.case_1()
Expand Down
29 changes: 29 additions & 0 deletions tests/data-files/file/catalog.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
{
"id": "examples",
"stac_extensions": [
"https://stac-extensions.github.io/file/v2.1.0/schema.json"
],
"type": "Catalog",
"title": "Example Catalog",
"stac_version": "1.0.0",
"description": "This catalog is a simple demonstration of an example catalog that is used to organize a hierarchy of collections and their items.",
"links": [
{
"rel": "root",
"href": "./catalog.json",
"type": "application/json"
},
{
"rel": "item",
"href": "./some-item.json",
"type": "application/json",
"title": "Some item",
"file:size": 8675309
},
{
"rel": "self",
"href": "https://raw.githubusercontent.com/radiantearth/stac-spec/v1.0.0/examples/catalog.json",
"type": "application/json"
}
]
}
Loading