Przesyłanie danych do GeoCatalog przy użyciu API zbiorczego przesyłania danych

W tym artykule przedstawiono sposób pozyskiwania wielu zasobów danych geoprzestrzennych jednocześnie w usłudze GeoCatalog przy użyciu interfejsu API pozyskiwania zbiorczego. Najpierw tworzymy i konfigurujemy źródło pozyskiwania geokatalogów. Utworzenie źródła pozyskiwania ustanawia bezpieczne połączenie między zasobem GeoCatalog i lokalizacją przechowywania istniejących danych geoprzestrzennych. Następnie utworzymy kolekcję SpatioTemporal Access Catalog (STAC) w ramach zasobu GeoCatalog w celu przechowywania pozyskiwanych danych. Na koniec użyjemy interfejsu API pozyskiwania zbiorczego, aby zainicjować przepływ pracy pozyskiwania. Po zakończeniu wykonywania tych kroków dane geoprzestrzenne są pozyskiwane i dostępne z interfejsów użytkownika oraz interfejsów API GeoKatalogu.

Wymagania wstępne

W ramach subskrypcji platformy Azure:

Zestaw danych geoprzestrzennych w kontenerze obiektów blob konta magazynowania.

Zasoby danych geoprzestrzennych (na przykład pliki GeoTIFF)
Skojarzone elementy STAC tworzą elementy STAC dla tych zasobów.
Kod JSON kolekcji STAC, który odwołuje się do wszystkich elementów STAC i zasobów danych geoprzestrzennych.

W swoim lokalnym lub programistycznym środowisku:

Środowisko Python z systemem Python 3.10 (lub nowszym).
Interfejs wiersza polecenia platformy Azure
Jesteś zalogowany do Azure

Komputer planetarny Microsoft Pro musi mieć dostęp do kontenera usługi Azure Blob Storage. W tym artykule utworzymy i użyjemy tymczasowego poświadczenia tokenu SAS w celu udzielenia tego dostępu. Alternatywnie możesz użyć tych przewodników, aby skonfigurować tożsamość zarządzaną lub zakodowane tokeny SAS.

API REST
SDK Python

Tworzenie źródła pozyskiwania

Utworzenie źródła pozyskiwania definiuje dla geokatalogu, które źródło służy do pozyskiwania danych geoprzestrzennych oraz mechanizm uwierzytelniania do użycia w przepływie pracy pozyskiwania.

Zainstaluj wymagane moduły Python przy użyciu pip

pip install pystac-client azure-identity requests azure-storage-blob pyyaml

Importowanie wymaganych modułów języka Python

import os
import requests
from azure.identity import AzureCliCredential
from datetime import datetime, timedelta, timezone
import azure.storage.blob
from urllib.parse import urlparse
import yaml

Ustawianie wymaganych stałych zgodnie ze środowiskiem

MPCPRO_APP_ID = "https://geocatalog.spatio.azure.com"
CONTAINER_URI = "<container_uri>" # The URI for the blob storage container housing your geospatial data
GEOCATALOG_URI = "<geocatalog uri>" # The URI for your GeoCatalog can be found in the Azure portal resource overview 
API_VERSION = "2026-04-15"

Tworzenie tokenu SAS

# Parse the container URL
parsed_url = urlparse(CONTAINER_URI)
account_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
account_name = parsed_url.netloc.split(".")[0]
container_name = parsed_url.path.lstrip("/")

credential = azure.identity.AzureCliCredential()
blob_service_client = azure.storage.blob.BlobServiceClient(
    account_url=account_url,
    credential=credential,
)

now = datetime.now(timezone.utc).replace(microsecond=0)
key = blob_service_client.get_user_delegation_key(
    key_start_time=now + timedelta(hours=-1),
    key_expiry_time=now + timedelta(hours=1),
)

sas_token = azure.storage.blob.generate_container_sas(
    account_name=account_name,
    container_name=container_name,
    user_delegation_key=key,
    permission=azure.storage.blob.ContainerSasPermissions(
        read=True,
        list=True,
    ),
    start=now + timedelta(hours=-1),
    expiry=now + timedelta(hours=1),
)

Uzyskaj token dostępu do GeoCatalog API

# Obtain an access token
credential = AzureCliCredential()
access_token = credential.get_token(f"{MPCPRO_APP_ID}/.default")

Tworzenie żądania POST przeznaczonego dla interfejsu API źródła pozyskiwania

# Payload for the POST request
payload = {
    "Kind": "SasToken",
    "connectionInfo": {
        "containerUrl": CONTAINER_URI,
        "sasToken": sas_token,
    },
}

Utwórz źródło pozyskiwania, wysyłając ładunek POST do punktu końcowego źródła pozyskiwania

# STAC Collection API endpoint
endpoint = f"{GEOCATALOG_URI}/inma/ingestion-sources"

# Make the POST request
response = requests.post(
    endpoint,
    json=payload,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

Weryfikowanie odpowiedzi

# Print the response
if response.status_code == 201:
    print("Ingestion source created successfully")
    ingestion_source_id = response.json().get("id") #saved for later to enable resource clean up
else:
    print(f"Failed to create ingestion: {response.text}")

Uwaga / Notatka

Wykonanie tych kroków więcej niż raz wstecz powoduje wyświetlenie odpowiedzi 409:

Container url <container uri> already contains a SAS token ingestion source with id <sas token id>

Interfejs API źródła pozyskiwania nie umożliwia utworzenia więcej niż jednego źródła pozyskiwania dla tego samego adresu URL kontenera. Aby uniknąć konfliktów, przed utworzeniem nowego należy wyczyścić wszystkie istniejące źródła pozyskiwania. Aby uzyskać szczegółowe informacje, zobacz Czyszczenie zasobów.

Utwórz kolekcję

Kolekcja STAC to kontener wysokiego poziomu dla elementów STAC i skojarzonych z nimi zasobów geoprzestrzennych. W tej sekcji utworzymy kolekcję STAC w naszym GeoKatalogu, aby pomieścić dane geoprzestrzenne, które będziemy pozyskiwać w następnej sekcji.

Importowanie wymaganych modułów

import os
import requests
import yaml
from pprint import pprint
from azure.identity import AzureCliCredential

Ustawianie wymaganych stałych zgodnie ze środowiskiem

MPCPRO_APP_ID = "https://geocatalog.spatio.azure.com"
GEOCATALOG_URI = "<geocatalog uri>" # The URI for your GeoCatalog can be found in the Azure portal resource overview 
API_VERSION = "2026-04-15"

COLLECTION_ID = "example-collection" #You can your own collection ID
COLLECTION_TITLE = "Example Collection" #You can your own collection title

Uzyskaj token dostępu do GeoCatalog API

# Obtain an access token
credential = AzureCliCredential()
access_token = credential.get_token(f"{MPCPRO_APP_ID}/.default")

Tworzenie podstawowej specyfikacji kolekcji STAC

collection = {
    "id": COLLECTION_ID,
    "type": "Collection",
    "title": COLLECTION_TITLE,
    "description": "An example collection",
    "license": "CC-BY-4.0",
    "extent": {
        "spatial": {"bbox": [[-180, -90, 180, 90]]},
        "temporal": {"interval": [["2018-01-01T00:00:00Z", "2018-12-31T23:59:59Z"]]},
    },
    "links": [],
    "stac_version": "1.0.0",
    "msft:short_description": "An example collection",
}

Uwaga / Notatka

Ta przykładowa specyfikacja kolekcji jest podstawowym przykładem kolekcji. Aby uzyskać więcej informacji na temat kolekcji STAC i standardu STAC open, zobacz przegląd STAC. Aby uzyskać więcej informacji na temat tworzenia kompletnej kolekcji STAC, zobacz Tworzenie kolekcji STAC.

Tworzenie nowej kolekcji przy użyciu interfejsu API kolekcji

response = requests.post(
    f"{GEOCATALOG_URI}/stac/collections",
    json=collection,
    headers={"Authorization": "Bearer " + access_token.token},
    params={"api-version": API_VERSION},
)

Weryfikowanie wyniku odpowiedzi

if response.status_code == 201:
    print("Collection created successfully")
    pprint(response.json())
else:
    print(f"Failed to create ingestion: {response.text}")

Tworzenie połączenia i uruchamianie przepływu pracy

W tym ostatnim kroku używamy interfejsu API do przyjmowania danych, aby zainicjować przepływ pracy do masowego przyjmowania danych.

Importowanie wymaganych modułów

import os
import requests
import yaml
from azure.identity import AzureCliCredential

Ustawianie wymaganych stałych zgodnie ze środowiskiem

MPCPRO_APP_ID = "https://geocatalog.spatio.azure.com"
GEOCATALOG_URI = "<geocatalog uri>" # The URI for your GeoCatalog can be found in the Azure portal resource overview 
API_VERSION = "2026-04-15"

COLLECTION_ID = "example-collection" #You can your own collection ID
catalog_href = "<catalog_href>" #The blob storage location of the STAC Catalog JSON file

skip_existing_items = False
keep_original_assets = False
timeout_seconds = 300

Uzyskaj token dostępu do GeoCatalog API

# Obtain an access token
credential = AzureCliCredential()
access_token = credential.get_token(f"{MPCPRO_APP_ID}/.default")

Tworzenie ładunku POST dla interfejsu API pozyskiwania zbiorczego

url = f"{GEOCATALOG_URI}/inma/collections/{COLLECTION_ID}/ingestions"
body = {
    "importType": "StaticCatalog",
    "sourceCatalogUrl": catalog_href,
    "skipExistingItems": skip_existing_items,
    "keepOriginalAssets": keep_original_assets,
}

Wyślij ładunek do interfejsu API pozyskiwania zbiorczego.

ing_response = requests.post(
    url,
    json=body,
    timeout=timeout_seconds,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

Sprawdź odpowiedź.

if ing_response.status_code == 201:
    print("Ingestion created successfully")
    ingestion_id = ing_response.json()["ingestionId"]
    print(f"Created ingestion with ID: {ingestion_id}")
else:
    print(f"Failed to create ingestion: {ing_response.text}")

Sprawdź stan przepływu pracy pozyskiwania.

runs_endpoint = (
    f"{geocatalog_url}/inma/collections/{collection_id}/ingestions/{ingestion_id}/runs"
)

wf_response = requests.post(
    runs_endpoint,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

if wf_response.status_code == 201:
    print("Workflow started successfully")
else:
    print(f"Failed to create ingestion run: {wf_response.text}")

Po zakończeniu przepływu pracy możesz wykonywać zapytania, pobierać lub wizualizować dane geoprzestrzenne przy użyciu geokatalogu STAC lub interfejsów API danych albo eksploratora danych. Jeśli wystąpią problemy, zapoznaj się z przewodnikiem rozwiązywania problemów lub listą kodów błędów przetwarzania.

Uprzątnij zasoby

Usuń źródło importu

del_is_endpoint = f"{GEOCATALOG_URI}/inma/ingestion-sources/{ingestion_source_id}"
del_is_response = requests.delete(
    del_is_endpoint,
    headers={"Authorization": f"Bearer {access_token.token}"},
    params={"api-version": API_VERSION},
)

if del_is_response.status_code == 200:
    print("Ingestion source deleted successfully")
else:
    print(f"Failed to delete ingestion source")

Zainstaluj zestaw SDK Python Planetary Computer Pro:

pip install azure-planetarycomputer azure-identity azure-storage-blob

Tworzenie źródła pozyskiwania

from azure.planetarycomputer import PlanetaryComputerProClient
from azure.identity import DefaultAzureCredential
from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse
import azure.storage.blob

client = PlanetaryComputerProClient(
    endpoint="<your-geocatalog-url>",
    credential=DefaultAzureCredential()
)

CONTAINER_URI = "<container_uri>"

# Generate SAS token using azure-storage-blob (same as REST API approach)
parsed_url = urlparse(CONTAINER_URI)
account_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
account_name = parsed_url.netloc.split(".")[0]
container_name = parsed_url.path.lstrip("/")

credential = DefaultAzureCredential()
blob_service_client = azure.storage.blob.BlobServiceClient(
    account_url=account_url,
    credential=credential,
)
now = datetime.now(timezone.utc).replace(microsecond=0)
key = blob_service_client.get_user_delegation_key(
    key_start_time=now + timedelta(hours=-1),
    key_expiry_time=now + timedelta(hours=1),
)
sas_token = azure.storage.blob.generate_container_sas(
    account_name=account_name,
    container_name=container_name,
    user_delegation_key=key,
    permission=azure.storage.blob.ContainerSasPermissions(read=True, list=True),
    start=now + timedelta(hours=-1),
    expiry=now + timedelta(hours=1),
)

# Create the ingestion source
ingestion_source = client.ingestion.create_source({
    "kind": "SasToken",
    "connectionInfo": {
        "containerUrl": CONTAINER_URI,
        "sasToken": sas_token,
    },
})
print(f"Ingestion source created: {ingestion_source['id']}")

Utwórz kolekcję

collection = {
    "id": "example-collection",
    "type": "Collection",
    "title": "Example Collection",
    "description": "An example collection",
    "license": "CC-BY-4.0",
    "extent": {
        "spatial": {"bbox": [[-180, -90, 180, 90]]},
        "temporal": {"interval": [["2018-01-01T00:00:00Z", "2018-12-31T23:59:59Z"]]},
    },
    "links": [],
    "stac_version": "1.0.0",
    "msft:short_description": "An example collection",
}

poller = client.stac.begin_create_collection(body=collection)
result = poller.result()
print(f"Collection created: {result}")

Tworzenie połączenia i uruchamianie przepływu pracy

COLLECTION_ID = "example-collection"
catalog_href = "<catalog_href>"  # Blob storage location of the STAC Catalog JSON file

ingestion = client.ingestion.create(COLLECTION_ID, {
    "importType": "StaticCatalog",
    "sourceCatalogUrl": catalog_href,
    "skipExistingItems": False,
    "keepOriginalAssets": False,
})
ingestion_id = ingestion.id
print(f"Created ingestion with ID: {ingestion_id}")

# Start the ingestion workflow
run = client.ingestion.create_run(COLLECTION_ID, ingestion_id)
print("Workflow started successfully")

Uprzątnij zasoby

client.ingestion.delete_source(ingestion_source["id"])
print("Ingestion source deleted successfully")

Aby uzyskać więcej informacji, zobacz dokumentację zestawu SDK Python.

Dalsze kroki

Po dodaniu kilku elementów należy skonfigurować dane na potrzeby wizualizacji.

Konfiguracja kolekcji w microsoft Planetary Computer Pro

Opinia

Czy ta strona była pomocna?

Ostatnia aktualizacja o 2026-06-03

Przesyłanie danych do GeoCatalog przy użyciu API zbiorczego przesyłania danych

Wymagania wstępne

Tworzenie źródła pozyskiwania

Utwórz kolekcję

Tworzenie połączenia i uruchamianie przepływu pracy

Uprzątnij zasoby

Dalsze kroki

Treści powiązane

Opinia

Dodatkowe zasoby