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
136 changes: 134 additions & 2 deletions src/pystac/link.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,18 +2,20 @@

import copy
import warnings
from typing import TYPE_CHECKING, Any, override
from typing import TYPE_CHECKING, Any, Self, override

from typing_extensions import deprecated

import pystac
from pystac.errors import STACError
from pystac.media_type import MediaType
from pystac.rel_type import RelType
from pystac.utils import make_absolute_href, make_posix_style
from pystac.utils import is_absolute_href, make_absolute_href, make_posix_style

from .reader import Reader

if TYPE_CHECKING:
from . import Catalog, Collection, Item
from .stac_object import STACObject

HIERARCHICAL_LINKS = [
Expand Down Expand Up @@ -46,10 +48,18 @@ def __init__(
self.method: str | None = method
self.headers: dict[str, str | list[str]] | None = headers
self.body: Any | None = body
extra_fields = kwargs.pop("extra_fields", None)
self.extra_fields: dict[str, Any] = kwargs
if extra_fields:
warnings.warn(
"Pass extra_fields entries as kwargs "
"instead of extra_fields as keyword argument."
)
self.extra_fields.update(extra_fields)

if isinstance(target, STACObject):
self._href: str | None = href or target.get_self_href()
self.title = title or getattr(target, "title", None)
self._target: STACObject | None = target
elif href and target:
raise ValueError("Both target and href were provided as strings")
Expand Down Expand Up @@ -82,6 +92,9 @@ def __getattribute__(self, name: str, /) -> Any:
else:
return super().__getattribute__(name)

def clone(self: Self) -> Self:
return copy.deepcopy(self)

@classmethod
def try_from(cls, data: dict[str, Any] | Link) -> Link:
if isinstance(data, Link):
Expand Down Expand Up @@ -120,6 +133,12 @@ def is_json(self) -> bool:
def get_href(self) -> str | None:
return self._href or self._target and self._target.get_self_href()

def get_absolute_href(self, start_href: str = "") -> str | None:
href = self.get_href()
if href is None:
return href
return make_absolute_href(href, start_href, start_is_dir=False)

def set_href(self, href: str) -> None:
self._href = href

Expand All @@ -133,6 +152,34 @@ def href(self) -> str | None:
def target(self) -> str | STACObject | None:
return self._target or self._href

@target.setter
@deprecated("target is deprecated, either use .set_href() or .set_target()")
def target(self, value: str | STACObject) -> None:
if isinstance(value, str):
warnings.warn(
"Setting Link.target to href is no longer supported pystac v2. "
"Assigning value to href instead."
)
self.set_href(value)
Comment thread
jsignell marked this conversation as resolved.
else:
self.set_target(value)
if href := value.get_self_href():
self.set_href(href)

@deprecated("use .get_href()")
def get_target_str(self) -> str | None:
"""Returns this link's target as a string.

If a string href was provided, returns that. If not, tries to resolve
the self link of the target object.
"""
return self.get_href()

@deprecated("use bool(link.get_href())")
def has_target_href(self) -> bool:
"""Returns true if this link has a string href in its target information."""
return bool(self.get_href())

def get_target(self, start_href: str | None, reader: Reader) -> STACObject:
from .stac_object import STACObject

Expand Down Expand Up @@ -183,6 +230,91 @@ def to_dict(self, transform_href: bool | None = None) -> dict[str, Any]:
data["body"] = self.body
return data

def resolve_stac_object(self, start_href: str = "") -> Link:
"""Resolves a STAC object from the HREF of this link, if the link is not
already resolved.

Args:
start_href : Optional string to put ahead of the href in this Link.

NOTE: This uses reader.DEFAULT_READER to read the HREF, if necessary.
"""
if self._target:
return self
elif self._href:
# If it's a relative link, base it off the parent.
target_href = self._href
if not is_absolute_href(target_href):
target_href = make_absolute_href(self._href, start_href=start_href)
try:
obj = pystac.read_file(target_href)
except Exception as e:
raise STACError(
f"HREF: '{target_href}' does not resolve to a STAC object"
) from e
self._target = obj
else:
raise ValueError("Cannot resolve STAC object without a target")

return self

@override
def __repr__(self) -> str:
return f"Link(rel={self.rel}, href={self._href})"

##### Convenience methods for Link creation #####
@classmethod
def root(cls: type[Self], c: Catalog) -> Self:
"""Creates a link to a root Catalog or Collection."""
return cls(RelType.ROOT, c, media_type=MediaType.JSON)

@classmethod
def parent(cls: type[Self], c: Catalog, title: str | None = None) -> Self:
"""Creates a link to a parent Catalog or Collection."""
return cls(RelType.PARENT, c, title=title, media_type=MediaType.JSON)

@classmethod
def collection(cls: type[Self], c: Collection) -> Self:
"""Creates a link to a Collection."""
return cls(RelType.COLLECTION, c, media_type=MediaType.JSON)

@classmethod
def self_href(cls: type[Self], href: str) -> Self:
"""Creates a self link to a file's location."""
return cls(RelType.SELF, href, media_type=MediaType.JSON)

@classmethod
def child(cls: type[Self], c: Catalog, title: str | None = None) -> Self:
"""Creates a link to a child Catalog or Collection."""
return cls(RelType.CHILD, c, title=title, media_type=MediaType.JSON)

@classmethod
def item(cls: type[Self], item: Item, title: str | None = None) -> Self:
"""Creates a link to an Item."""
return cls(RelType.ITEM, item, title=title, media_type=MediaType.GEOJSON)

@classmethod
def derived_from(
cls: type[Self], item: Item | str, title: str | None = None
) -> Self:
"""Creates a link to a derived_from Item."""
return cls(
RelType.DERIVED_FROM,
item,
title=title,
media_type=MediaType.JSON,
)

@classmethod
def canonical(
cls: type[Self],
item_or_collection: Item | Collection,
title: str | None = None,
) -> Self:
"""Creates a canonical link to an Item or Collection."""
return cls(
RelType.CANONICAL,
item_or_collection,
title=title,
media_type=MediaType.JSON,
)
7 changes: 7 additions & 0 deletions tests/v1/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,13 @@ def collection() -> Catalog:
return Collection("test-collection", "A test collection", ARBITRARY_EXTENT)


@pytest.fixture
def self_link_collection() -> Catalog:
c = Collection("test-collection", "A test collection", ARBITRARY_EXTENT)
c.set_self_href("file:///a/real/url.json")
return c


@pytest.fixture
def multi_extent_collection() -> Collection:
# TODO this code is repeated many times; refactor to use this fixture
Expand Down
10 changes: 10 additions & 0 deletions tests/v1/posix_paths/test_posix_paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,16 @@ def test_create_item_containing_posix_hrefs(tmp_path: Path) -> None:
pystac.read_file(collection_href)


def test_posix_self_link_added_from_file(tmp_path: Path) -> None:
href = get_data_file("item/sample-item.json")
item = pystac.Item.from_file(href)
item.links.append(pystac.Link(rel="self", href=href))
check_link(item.get_single_link(rel="self"))
item2 = pystac.read_file(href)
item2.links.append(pystac.Link(rel="self", href=href))
check_link(item2.get_single_link(rel="self"))


@pytest.mark.skipif(os.name != "nt", reason="windows only test")
def test_posix_self_link_from_absolute_href(tmp_path: Path) -> None:
# Check that we convert to a windows style (\\) absolute href to posix style
Expand Down
Loading