Module stac_fastapi.core.validate¶
STAC validation module.
Provides validation for STAC items and collections using multiple validation backends: - Pydantic validation (always enabled via FastAPI/stac_pydantic) - Fast JSON Schema compiled validation (ultra-fast, sequential STAC validation)
Variables¶
fastjsonschema
fv_module
get_validator
logger
Functions¶
async_validate_batch_with_stac_validator¶
def async_validate_batch_with_stac_validator(
items: list[dict]
) -> tuple[list[dict], dict[str, list[str]]]
Asynchronously validate a batch of STAC items.
Offloads the CPU-bound validation loop to a separate thread to prevent blocking the FastAPI asyncio event loop.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| items | None | List of STAC item dictionaries to validate. | None |
Returns:
| Type | Description |
|---|---|
| None | Tuple of (valid_items_list, invalid_items_dict) where invalid_items_dict maps error messages to lists of affected item IDs. |
async_validate_stac¶
def async_validate_stac(
stac_data: dict | stac_pydantic.item.Item | stac_pydantic.collection.Collection,
pydantic_model: type[stac_pydantic.item.Item] | type[stac_pydantic.collection.Collection] = <class 'stac_pydantic.item.Item'>
) -> stac_pydantic.item.Item | stac_pydantic.collection.Collection
Asynchronous wrapper for validate_stac.
Offloads the validation to a separate thread to prevent blocking the FastAPI asyncio event loop during API requests.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| stac_data | None | STAC data as dict or Pydantic model. | None |
| pydantic_model | None | The Pydantic model class to use for validation (Item or Collection). | None |
Returns:
| Type | Description |
|---|---|
| None | Validated STAC object (Item or Collection). |
Raises:
| Type | Description |
|---|---|
| ValueError | If validation fails. |
batch_validate_topology¶
def batch_validate_topology(
features: list[dict]
) -> tuple[list[dict], dict[str, list[str]]]
Synchronize batch topology validation suitable for thread pool execution.
Processes an array of features, returning the valid ones alongside a dictionary of caught topology errors. Designed to be offloaded to a background thread via asyncio.to_thread to prevent blocking the FastAPI event loop during CPU-bound geometry validation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| features | None | List of feature dictionaries to validate. | None |
Returns:
| Type | Description |
|---|---|
| None | Tuple of (valid_features, topology_errors) where topology_errors maps error messages to lists of affected item IDs. |
validate_batch_with_stac_validator¶
def validate_batch_with_stac_validator(
items: list[dict]
) -> tuple[list[dict], dict[str, list[str]]]
Validate a batch of STAC items using compiled fastjsonschema functions.
Performs ultra-fast sequential validation in memory. Bypasses multiprocessing to save massive amounts of RAM and Inter-Process Communication (IPC) CPU overhead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| items | None | List of STAC item dictionaries to validate. | None |
Returns:
| Type | Description |
|---|---|
| None | Tuple of (valid_items_list, invalid_items_dict) where invalid_items_dict maps error messages to lists of affected item IDs. |
validate_datetime_range¶
def validate_datetime_range(
item: dict
) -> None
Validate datetime range fields in a STAC item.
Validates STAC spec requirements: - If datetime is null, both start_datetime and end_datetime must be set - If start_datetime is set, end_datetime must also be set (and vice versa) - start_datetime must be <= end_datetime - If datetime is set, it must fall within start/end range
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| item | dict | STAC item to validate. | None |
Raises:
| Type | Description |
|---|---|
| ValueError | If datetime range validation fails. |
validate_geometry_bounds¶
def validate_geometry_bounds(
coords: list
) -> None
Recursively validate that all coordinate pairs fall within WGS84 limits.
Traverses nested coordinate lists to enforce global bounds (±180° lon, ±90° lat) across all geometry types (Point, LineString, Polygon, MultiPolygon, etc.).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| coords | None | Coordinate list (may be nested). | None |
Raises:
| Type | Description |
|---|---|
| ValueError | If any coordinate pair falls outside WGS84 bounds. |
validate_item_topology_lightweight¶
def validate_item_topology_lightweight(
item_dict: dict
) -> None
Lightweight validation to enforce global coordinate boundaries and antimeridian checks.
Validates all geometry types for WGS84 bounds compliance, and checks for improper antimeridian line skips in Polygon and MultiPolygon rings. Polygon rings are also checked against a configurable vertex limit (default 5000) to prevent DoS attacks with pathologically complex geometries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| item_dict | None | The STAC item dictionary to validate. | None |
Raises:
| Type | Description |
|---|---|
| ValueError | If geometry is invalid, out of bounds, crosses antimeridian improperly, or exceeds the vertex limit. |
validate_stac¶
def validate_stac(
stac_data: dict | stac_pydantic.item.Item | stac_pydantic.collection.Collection,
pydantic_model: type[stac_pydantic.item.Item] | type[stac_pydantic.collection.Collection] = <class 'stac_pydantic.item.Item'>
) -> stac_pydantic.item.Item | stac_pydantic.collection.Collection
Validate a single STAC item or collection using the optional STAC validator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| stac_data | None | STAC data as a raw dict or parsed Pydantic model object. | None |
| pydantic_model | None | The Pydantic model class to use for validation (Item or Collection). | None |
Returns:
| Type | Description |
|---|---|
| None | Validated STAC object (Item or Collection). |
Raises:
| Type | Description |
|---|---|
| ValueError | If strict STAC validation fails. |