Perform item transactions¶
These recipes assume the package was installed with the PySTAC extra:
python -m pip install "stac-transaction-api-client[pystac]"
Create a client¶
For a public API:
from stac_transaction_api_client import Client
client = Client(base_url="https://stac.example.com")
For a bearer-token protected API:
from stac_transaction_api_client import AuthenticatedClient
client = AuthenticatedClient(
base_url="https://stac.example.com",
token="your-access-token",
)
You can also pass headers, cookies, timeout, verify_ssl,
follow_redirects, and additional httpx_args when constructing either client.
Create one item¶
Pass a pystac.Item to post_feature:
from datetime import datetime, timezone
from pystac import Item
from stac_transaction_api_client.api.transaction import post_feature
item = Item(
id="example-item",
geometry={"type": "Point", "coordinates": [12.5, 41.9]},
bbox=[12.5, 41.9, 12.5, 41.9],
datetime=datetime.now(timezone.utc),
properties={},
)
response = post_feature.sync_detailed(
"example-collection",
client=client,
body=item,
)
print(response.status_code)
print(response.parsed)
Use post_feature.sync(...) if you only need the parsed result.
For bulk creation, pass a pystac.ItemCollection as body.
Retrieve an item¶
get_feature exposes detailed variants only:
from stac_transaction_api_client.api.transaction import get_feature
response = get_feature.sync_detailed(
"example-collection",
"example-item",
client=client,
)
item = response.parsed
if item is not None:
print(item.to_dict())
Use the detailed response when you need the ETag:
etag = response.headers.get("ETag")
Replace an item with PUT¶
update_feature requires the complete pystac.Item and an if_match string.
Fetch the current resource first, then update it with the returned ETag:
from stac_transaction_api_client.api.transaction import get_feature, update_feature
current = get_feature.sync_detailed(
"example-collection",
"example-item",
client=client,
)
if current.parsed is None:
raise RuntimeError("item not found")
etag = current.headers.get("ETag")
if etag is None:
raise RuntimeError("service did not return an ETag")
current.parsed.properties["processing:state"] = "ready"
updated = update_feature.sync_detailed(
"example-collection",
"example-item",
client=client,
body=current.parsed,
if_match=etag,
)
If another writer changed the resource first, the server can reject the request
with 412 Precondition Failed. Fetch the item again to obtain its new state and
ETag before deciding whether to retry.
Patch an item¶
patch_feature maps to the HTTP PATCH endpoint and accepts if_match
optionally:
from stac_transaction_api_client.api.transaction import patch_feature
patched = patch_feature.sync_detailed(
"example-collection",
"example-item",
client=client,
body={"properties": {"quality": "updated"}},
if_match=etag,
)
The partial mapping is sent as the JSON object required by the Transaction
Extension. A complete pystac.Item is also accepted when appropriate.
Delete an item¶
delete_feature requires an if_match value:
from stac_transaction_api_client.api.transaction import delete_feature, get_feature
current = get_feature.sync_detailed(
"example-collection",
"example-item",
client=client,
)
etag = current.headers.get("ETag")
if etag is None:
raise RuntimeError("service did not return an ETag")
deleted = delete_feature.sync_detailed(
"example-collection",
"example-item",
client=client,
if_match=etag,
)
print(deleted.status_code)
Use asynchronous calls¶
Each mutation module exposes asyncio_detailed(...) and asyncio(...) variants.
get_feature exposes asyncio_detailed(...).
import asyncio
from stac_transaction_api_client import Client
from stac_transaction_api_client.api.transaction import get_feature
async def main() -> None:
async with Client(base_url="https://stac.example.com") as client:
response = await get_feature.asyncio_detailed(
"example-collection",
"example-item",
client=client,
)
print(response.status_code)
asyncio.run(main())
Inspect the raw HTTP response¶
Use a *_detailed function whenever you need more than the parsed object. It
returns Response[T] with:
status_code: anhttp.HTTPStatusvalue;headers: the response headers, including ETag when supplied by the server;content: raw response bytes;parsed: the parsed PySTAC Item, error model, orNone, depending on the endpoint and status.