Ingérer des données dans GeoCatalog avec l’API d’ingestion en bloc

Cet article explique comment ingérer plusieurs ressources de données géospatiales à la fois dans un GeoCatalog à l’aide de l’API d’ingestion en bloc. Tout d’abord, nous créons et configurons une source d’ingestion GeoCatalog. La création d’une source d’ingestion établit une connexion sécurisée entre une ressource GeoCatalog et l’emplacement de stockage de vos données géospatiales existantes. Ensuite, nous créons une collection SpatioTemporal Access Catalog (STAC) au sein de votre ressource GeoCatalog pour stocker les données ingérées. Enfin, nous utilisons l’API d’ingestion en bloc pour lancer le flux de travail d’ingestion. Une fois ces étapes terminées, vos données géospatiales sont ingérées et accessibles à partir des interfaces et des API GeoCatalog.

Conditions préalables

Dans votre abonnement Azure :

Un jeu de données géospatial dans le conteneur blob de votre compte de stockage :

Ressources de données géospatiales (par exemple, fichiers GeoTIFF)
Les éléments STAC associés créent des éléments STAC pour ces ressources.
Un fichier JSON de collection STAC qui référence tous les éléments STAC et les ressources de données géospatiales.

Dans votre environnement local/développement :

Environnement Python exécutant Python 3.10 (ou version ultérieure).
Azure CLI
Vous êtes connecté à Azure

Microsoft Planetary Computer Pro doit avoir accès au conteneur Stockage Blob Azure. Dans cet article, nous créons et utilisons un jeton SAS temporaire pour accorder cet accès. Vous pouvez également utiliser ces guides pour configurer l’identité managée ou les jetons SAP codés en dur.

REST API
Kit de développement logiciel (SDK) Python

Créer une source d’ingestion

La création d’une source d’ingestion définit pour GeoCatalog la source à partir de laquelle ingérer des données géospatiales et le mécanisme d’informations d’identification à utiliser dans le flux de travail d’ingestion.

Installer les modules Python requis à l’aide de pip

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

Importer les modules Python requis

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

Définir les constantes requises en fonction de votre environnement

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"

Créer le jeton SAP

# 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),
)

Obtenir le jeton d’accès de l’API GeoCatalog

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

Créer une charge POST pour l’API source d’entrée

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

Créez la source d’ingestion en envoyant la charge utile POST au point de terminaison source d’ingestion

# 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},
)

Vérifier la réponse

# 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}")

Remarque

L’exécution de ces étapes plusieurs fois de retour en arrière entraîne une réponse 409 :

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

L’API source d’ingestion ne vous permet pas de créer plusieurs sources d’ingestion pour la même URL de conteneur. Pour éviter les conflits, veillez à nettoyer les sources d’ingestion existantes avant de en créer une nouvelle. Pour plus d’informations, consultez Nettoyer les ressources.

Créer une collection

Une collection STAC est le conteneur de haut niveau pour les éléments STAC et leurs ressources géospatiales associées. Dans cette section, nous créons une collection STAC dans notre GeoCatalog pour héberger les données géospatiales que nous ingérerons dans la section suivante.

Importez les modules nécessaires

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

Définir les constantes requises en fonction de votre environnement

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

Obtenir le jeton d’accès de l’API GeoCatalog

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

Créer une spécification de collection STAC de base

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",
}

Remarque

Cet exemple de spécification de collection est un exemple de base d’une collection. Pour plus d’informations sur les collections STAC et la norme ouverte STAC, consultez la vue d’ensemble de STAC. Pour plus d’informations sur la création d’une collection STAC complète, consultez Créer une collection STAC.

Créer la collection avec l’API Collection

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

Vérifier le résultat de la réponse

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

Créer une connexion et exécuter un flux de travail

Dans cette dernière étape, nous utilisons l’API d’ingestion pour lancer un flux de travail d’ingestion en bloc.

Importer les modules requis

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

Définir les constantes requises en fonction de votre environnement

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

Obtenir le jeton d’accès de l’API GeoCatalog

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

Créer la charge utile POST pour l’API d’ingestion en bloc

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

Envoyez la charge utile à l’API d’ingestion en bloc.

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

Vérifiez la réponse.

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}")

Vérifiez l’état du processus d’ingestion.

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}")

Une fois le flux de travail terminé, vous pouvez interroger, récupérer ou visualiser vos données géospatiales à l’aide des API STAC ou données GeoCatalog, ou avec l’Explorateur de données. Si vous rencontrez des problèmes, reportez-vous au guide de résolution des problèmes ou à la liste des codes d’erreur d’ingestion.

Nettoyer les ressources

Supprimer la source d’ingestion

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")

Installez le Kit de développement logiciel (SDK) Planetary Computer Pro Python :

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

Créer une source d’ingestion

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']}")

Créer une collection

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}")

Créer une connexion et exécuter un flux de travail

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")

Nettoyer les ressources

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

Pour plus d’informations, consultez la référence Python SDK.

Étapes suivantes

Maintenant que vous avez ajouté quelques éléments, vous devez configurer les données pour la visualisation.

Configuration du regroupement dans Microsoft Planetary Computer Pro

Commentaires

Est-ce que cette page vous a été utile?

Dernière mise à jour : 2026-06-03

Ingérer des données dans GeoCatalog avec l’API d’ingestion en bloc

Conditions préalables

Créer une source d’ingestion

Créer une collection

Créer une connexion et exécuter un flux de travail

Nettoyer les ressources

Étapes suivantes

Contenu connexe

Commentaires

Ressources supplémentaires