satorbis_kit.pgstac.models.config module¶

Ingestion configuration model for STAC workflows.

class satorbis_kit.pgstac.models.config.IngestionConfig(*args: Any, **kwargs: Any)[source]¶

Bases: BaseModel

Configuration for STAC ingestion workflow.

This model represents the configuration sent to Airflow DAG for STAC ingestion. It includes all necessary parameters for raster stacking and STAC catalog ingestion.

raster_s3_urls¶

List of S3 URLs to raster files

Type:: List[str]

collection¶

STAC collection name

Type:: str

costing_labels¶

Costing labels for tracking (always included)

Type:: Dict[str, str]

ingestion_batch_size¶

Batch size for ingestion (default: 100)

Type:: int | None

parameters¶

Optional COG conversion parameters (default: {})

Type:: Dict[str, Any] | None

Example

>>> config = IngestionConfig(
...     raster_s3_urls=["s3://bucket/TEST_COL/path/20240401.tif"],
...     collection="TEST_COL",
...     costing_labels=DEFAULT_COSTING_LABELS,
... )
>>> payload = config.to_payload()

collection: str¶

costing_labels: Dict[str, str]¶

raster_s3_urls: List[str]¶

to_payload() → Dict[str, Any][source]¶

Convert to Airflow DAG payload format.

Transforms the configuration into the exact format expected by the Airflow stac_item_batch_creation_dag. The configuration is wrapped in a “conf” key as per Airflow DAG trigger requirements.

Returns:

{

“conf”: {: “costing_labels”: {…}, “raster_s3_urls”: […], “collection”: “…”, “ingestion_batch_size”: 100, “parameters”: {…}

}

Return type:

Dictionary payload ready for Airflow API with structure

Example

>>> config = IngestionConfig(...)
>>> payload = config.to_payload()
>>> # payload is ready to send to Airflow