Skip to content

pgstac integration

stac_geoparquet.pgstac_reader has some helpers for working with items coming from a pgstac.items table. It takes care of

  • Rehydrating the dehydrated items
  • Partitioning by time
  • Injecting dynamic links and assets from a STAC API

stac_geoparquet.pgstac_reader.CollectionConfig dataclass

Additional collection-based configuration to inject, matching the dynamic properties from the API.

collection property 

collection: Collection

collection_id instance-attribute 

collection_id: str

partition_frequency class-attribute instance-attribute 

partition_frequency: str | None = None

render_config class-attribute instance-attribute 

render_config: str | None = None

should_inject_dynamic_properties class-attribute instance-attribute 

should_inject_dynamic_properties: bool = True

stac_api class-attribute instance-attribute 

stac_api: str = 'https://planetarycomputer.microsoft.com/api/stac/v1'

__init__ 

__init__(
    collection_id: str,
    partition_frequency: str | None = None,
    stac_api: str = "https://planetarycomputer.microsoft.com/api/stac/v1",
    should_inject_dynamic_properties: bool = True,
    render_config: str | None = None,
) -> None

__post_init__ 

__post_init__() -> None

export_collection 

export_collection(
    conninfo: str,
    output_protocol: str,
    output_path: str,
    storage_options: dict[str, Any],
    rewrite: bool = False,
    skip_empty_partitions: bool = False,
) -> list[str | None]

export_partition 

export_partition(
    conninfo: str,
    query: str,
    output_protocol: str,
    output_path: str,
    storage_options: dict[str, Any] | None = None,
    rewrite: bool = False,
    skip_empty_partitions: bool = False,
) -> str | None

export_partition_for_endpoints 

export_partition_for_endpoints(
    endpoints: tuple[datetime, datetime],
    conninfo: str,
    output_protocol: str,
    output_path: str,
    storage_options: dict[str, Any],
    part_number: int | None = None,
    total: int | None = None,
    rewrite: bool = False,
    skip_empty_partitions: bool = False,
) -> str | None

Export results for a pair of endpoints.

generate_endpoints 

generate_endpoints(
    since: datetime | None = None,
) -> list[tuple[datetime, datetime]]

inject_assets 

inject_assets(item: dict[str, Any]) -> None

inject_links(item: dict[str, Any]) -> None

make_pgstac_items 

make_pgstac_items(
    records: list[tuple[str, str, str, datetime, datetime, dict[str, Any]]],
    base_item: dict[str, Any],
) -> list[dict[str, Any]]

Make STAC items out of pgstac records.

Parameters:

  • records (list[tuple[str, str, str, datetime, datetime, dict[str, Any]]]) –

    list[tuple] The dehydrated records from pgstac.items table.

  • base_item (dict[str, Any]) –

    dict[str, Any] The base item from the collection_base_item pgstac function for this collection. Used for rehydration