Tutorial: Criar um mapa em tempo real usando APIs REST e Python

Os Mapas do Fabric podem visualizar dados geoespaciais em tempo real ao se conectar a conjuntos de dados do Eventhouse que são atualizados continuamente por meio da ingestão do Eventstream.

Ao contrário de cenários estáticos que usam arquivos armazenados em um Lakehouse, este tutorial demonstra uma arquitetura controlada por eventos e streaming em que:

• Os eventos são importados para um Eventhouse
• Os dados são consultados usando a KQL (Linguagem de Consulta Kusto)
• Um mapa é atualizado dinamicamente à medida que novos dados chegam

Este tutorial se concentra na automatização do fluxo de trabalho de ponta a ponta usando APIs REST do Fabric e Python, para que você possa provisionar recursos e configurar uma experiência de mapa em tempo real programaticamente. Para cenários de dados estáticos usando arquivos lakehouse, consulte Criar um mapa estático usando APIs REST e Python.

Neste tutorial, você aprenderá a criar e automatizar uma solução geoespacial em tempo real em Microsoft Fabric usando Eventstream, Eventhouse e KQL.

Para simular o streaming contínuo e assistir à atualização do mapa quase em tempo real, conclua este tutorial primeiro e continue com o acompanhamento Tutorial: simular a ingestão de dados em tempo real para um mapa usando APIs REST e Python, que se baseia diretamente na casa de eventos, no fluxo de eventos, na função KQL e no mapa que você cria aqui.

Visão geral do cenário: Acompanhamento de ativos em tempo real

Este tutorial baseia-se em um cenário de rastreamento de ativos em tempo real, semelhante ao cenário de rastreamento de frotas usado no tutorial original do Fabric Maps Tutorial: criar o roteamento de ordens de serviço em tempo real com o Fabric Maps.

Neste cenário:

• Os veículos emitem periodicamente atualizações de localização
• Eventos de localização são ingeridos em uma casa de eventos
• Um mapa exibe as últimas posições e atualizações do veículo automaticamente à medida que novos eventos chegam

Esse padrão representa casos comuns de uso operacional em tempo real, como:

• Acompanhamento de frota
• Expedição de ordens de serviço
• Monitoramento de ativos e equipamentos

Microsoft Fabric usa Eventstream e Eventhouse para ingerir, processar e analisar dados de streaming quase em tempo real, possibilitando visualizar dados operacionais dinâmicos diretamente em um mapa.

Este tutorial segue um padrão de automação comum em Fabric: criar uma infraestrutura → fluxo de ingestão → validar a ingestão → mapa de renderização.

Prerequisites

• Python 3.10 ou posterior
• CLI do Azure
• ID do espaço de trabalho Fabric
• Permissões para chamar APIs REST do Fabric, como:
  • Item.ReadWrite.All

Autenticação

Este tutorial usa DefaultAzureCredential, que pode se autenticar usando várias fontes de credenciais locais/dev. Para leitores iniciantes, a abordagem mais simples é a entrada na CLI do Azure.

Autenticar localmente (recomendado para a primeira execução)

1. Abra um terminal.
2. Executar:

az login

DefaultAzureCredential pode usar sua identidade autenticada para adquirir tokens de acesso para:

• APIs REST do Fabric (recurso: https://api.fabric.microsoft.com/.default)
• Consultas de plano de dados do Kusto (KQL) no eventhouse (recurso: https://api.kusto.windows.net/.default)
• Power BI / Fabric endpoints REST usados para consultar operações de longa duração (recurso: https://analysis.windows.net/powerbi/api/.default)

Entrar no Microsoft Fabric (recomendado)

Antes de executar este tutorial, recomendamos entrar no Microsoft Fabric pelo menos uma vez:

https://app.fabric.microsoft.com

Fazer login garante que sua identidade de Fabric, a associação de função no workspace e as atribuições de capacidade sejam totalmente provisionadas antes de adquirir um token de acesso do Microsoft Entra de forma programática.

Esta etapa será especialmente útil se:

• Você é novo no Microsoft Fabric
• O workspace foi criado recentemente
• Sua atribuição de função foi adicionada recentemente

Criar o arquivo de dados de semente (dados iniciais do mapa)

Para garantir que o mapa exiba dados imediatamente após o provisionamento, o script envia um pequeno conjunto de eventos de semente para o fluxo de eventos.

1. Crie um novo arquivo no mesmo diretório que o script Python: vehicle_locations_seed.csv
2. Cole o seguinte conteúdo:

VehicleId,Latitude,Longitude,EventTime
V-001,47.6101,-122.3344,2026-01-01T10:00:00Z
V-002,47.6150,-122.3200,2026-01-01T10:00:00Z
V-003,47.6205,-122.3493,2026-01-01T10:00:00Z
V-004,47.6050,-122.3300,2026-01-01T10:00:00Z

Etapa 1 – Criar um novo arquivo de projeto do Python

Nesta etapa, você criará um arquivo de Python em branco que você compila seção a seção.

Crie um novo arquivo chamado:

create_realtime_map.py

Abra o arquivo em seu editor.

Etapa 2 – Instalar bibliotecas necessárias e adicionar instruções de importação necessárias

Nesta etapa, você instala as dependências e adiciona as importações que o script usa.

Instalar bibliotecas necessárias

Na janela do terminal que você acabou de abrir, execute o seguinte comando:

pip install httpx azure-identity azure-eventhub

Para que serve cada biblioteca

• httpx: faz solicitações HTTP para as APIs REST do Fabric.
• azure-identity: fornece DefaultAzureCredential para autenticação do Microsoft Entra.
• azure-eventhub: envia eventos iniciais para o endpoint compatível com o Event Hub do Eventstream para preencher o eventhouse.

Adicionar instruções de importação ao arquivo .py

Na parte superior do create_realtime_map.py, adicione:

import base64
import csv
import json
import os
import time
import uuid

import httpx
from azure.eventhub import EventData, EventHubProducerClient
from azure.eventhub.exceptions import EventHubError
from azure.identity import DefaultAzureCredential

Etapa 3 – Adicionar uma seção de configuração

Nesta etapa, você define as variáveis que seu aplicativo usa, incluindo a ID do workspace e os nomes de recursos.

Centralizar a configuração em uma única Config classe , em vez de dispersar valores codificados em funções , oferece três vantagens concretas:

• Portabilidade do ambiente: IDs de workspace, nomes de recursos e outras configurações residem em um só lugar, para que você possa executar novamente o script em um workspace ou computador diferente alterando algumas linhas (ou uma variável de ambiente) em vez de procurar pelo código.
• Assinaturas de função mais limpas: as funções de etapa aceitam um único objeto cfg em vez de longas listas de parâmetros, o que mantém a orquestração em main() fácil de entender.
• Tratamento de segredos mais seguros: valores confidenciais, como a ID do workspace, são carregados de variáveis de ambiente, para que nunca sejam confirmados junto com o script.

Adicione o seguinte abaixo das instruções de importação :

# =========================================================
# Configuration (centralized)
# =========================================================

class Config:
    """
    Central configuration: workspace ID, resource display names, and
    ingestion settings. A single instance is built in main() and passed
    to each step function.
    """
    def __init__(self):
        # Workspace
        self.workspace_id = os.environ.get("FABRIC_WORKSPACE_ID", "")
        if not self.workspace_id:
            raise RuntimeError("Set FABRIC_WORKSPACE_ID environment variable before running the script.")

        # Resource display names / descriptions
        self.eventhouse_display_name = "eh_realtime_locations"
        self.eventhouse_description = "Stores streaming location events for a Fabric Maps real-time tutorial"

        self.eventhouse_table_name = "VehicleLocations"
        self.kql_function_name = "LatestVehicleLocations"

        self.eventstream_display_name = "es_realtime_locations"
        self.eventstream_description = "Streams events into an eventhouse table (created via Eventstream REST API)"

        self.map_display_name = "My Real-Time Fabric Map"
        self.map_description = "Created using Fabric Maps REST API (Eventhouse + Eventstream + Kusto function)"

        # Map refresh
        self.refresh_interval_ms = 5000

        # Seed data (initial map data)
        self.seed_csv_path = os.path.join(os.path.dirname(__file__), "vehicle_locations_seed.csv")
        
        # Will be provided interactively after eventstream is created
        self.eventhub_connection_string = os.environ.get("EVENTHUB_CONNECTION_STRING", "")

Definir a ID do workspace usando uma variável de ambiente

Em vez de fixar o ID do espaço de trabalho diretamente no script, este tutorial lê esse valor de uma variável de ambiente. Isso mantém valores específicos do ambiente fora do código-fonte e permite reutilizar o script em workspaces ou computadores sem editá-lo.

Antes de executar o script, crie uma variável de ambiente chamada FABRIC_WORKSPACE_ID.

Definir a variável de ambiente em Windows

No Windows, você pode definir a variável em qualquer terminal que ofereça suporte a variáveis de ambiente — PowerShell, Windows PowerShell, as janelas do PowerShell ou do Prompt de Comando integradas ao Visual Studio e ao Visual Studio Code, Terminal do Windows e a maioria dos outros shells.

Execute o seguinte no PowerShell ou no terminal integrado do VS Code:

$env:FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Para confirmar se a variável está definida:

echo $env:FABRIC_WORKSPACE_ID

Isso define a variável somente para a sessão do terminal atual.

Definir uma variável de ambiente persistente (Windows)

Para disponibilizar a variável em sessões futuras, use um dos seguintes:

• PowerShell (one-liner): Executar setx FABRIC_WORKSPACE_ID "<WORKSPACE_ID>". O comando setx grava no ambiente de usuário, mas não atualiza o terminal atual — feche e reabra o terminal (ou abra um novo) antes de executar o script.
• GUI:
  1. Abrir propriedades do sistema.
  2. Selecione Configurações avançadas do sistema.
  3. Escolha variáveis de ambiente.
  4. Em Variáveis de usuário, selecione Novo.
  5. Entrar:
     • Nome: FABRIC_WORKSPACE_ID
     • Valor: ID do seu espaço de trabalho
  6. Selecione OK para salvar.
  7. Feche e reabra o terminal antes de executar o script novamente.

Definir a variável de ambiente no macOS ou linux

No macOS e no Linux, você pode definir a variável de qualquer shell que dê suporte a export — Bash, Zsh (o padrão no macOS moderno), o Fish (com uma sintaxe ligeiramente diferente) e os terminais integrados em Visual Studio Code e outros editores.

Executar:

export FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Para confirmar se a variável está definida:

echo $FABRIC_WORKSPACE_ID

Isso define a variável somente para a sessão do shell atual.

Definir uma variável de ambiente persistente (macOS ou Linux)

Para disponibilizar a variável em sessões futuras, adicione a export linha ao seu perfil de shell:

• Zsh (padrão no macOS): ~/.zshrc
• Bash: ~/.bashrc (Linux) ou ~/.bash_profile (macOS)
• Peixe: executar set -Ux FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" em vez de editar um arquivo

Depois de atualizar o perfil, abra um novo terminal ou execute source ~/.zshrc (ou o arquivo apropriado) para que a alteração entre em vigor.

Etapa 4 – Adicionar funções auxiliares

Nesta etapa, você isola aspectos transversais — autenticação, construção de cabeçalhos, sondagem de operações de longa duração e lógica de nova tentativa — em um pequeno conjunto de funções auxiliares reutilizáveis que cada função de etapa pode chamar.

Centralizar essas preocupações em funções auxiliares — em vez de inseri-las diretamente em cada ponto de chamada — oferece três vantagens concretas:

• Fonte única da verdade para aspectos transversais: autenticação, cabeçalhos e sondagem de LRO são necessários em praticamente todas as chamadas de API. Centralizá-los mantém a função de cada etapa focada em seu próprio recurso, em vez de reimplementar a obtenção de token e a lógica de repetição.
• Resiliência sem bagunça: os auxiliares absorvem condições transitórias — provisionamento assíncrono, atraso na propagação do backend, falhas de envio passíveis de repetição — para que as funções de etapas permaneçam curtas e possam ser lidas como uma lista de verificação.
• Mais fácil de ensinar e modificar: cada auxiliar é introduzido uma vez e reutilizado. Se o Fabric alterar um padrão de LRO ou um escopo de autenticação, você corrige isso em um só lugar.

Os auxiliares que você adiciona nesta etapa são:

• Auxiliares de autenticação: criar cabeçalhos para APIs REST do Fabric (e endpoints de LRO do cluster do Power BI)
• FabricClient: wrapper leve para chamadas de API consistentes
• Manipulador de LRO: consultar operações de longa duração usando Location / x-ms-operation-id / Retry-After, incluindo respostas 200com-Running, endpoints de cluster do Power BI e cargas de conclusão apenas com status (resolvido por displayName)
• Auxiliar de carga útil de definição: codificar map.json em base64 para definições embutidas
• Assistente de conexão do Eventstream: solicita a string de conexão do ponto de extremidade personalizado
• Auxiliar de semente: envia eventos iniciais com lógica de repetição para garantir que a ingestão seja bem-sucedida
• Auxiliar de prontidão do banco de dados KQL: aguarda até que o banco de dados KQL esteja disponível para o Mapas do Fabric

Criar funções auxiliares de autenticação

Cada chamada à API REST do Fabric feita neste tutorial inclui um token de acesso do Microsoft Entra (token bearer) no cabeçalho Authorization. Em vez de adquirir tokens de forma ad hoc, esta etapa encapsula DefaultAzureCredential em um pequeno TokenProvider e expõe um construtor de cabeçalhos específico para cada público para cada família de endpoints que o script chama.

Centralizar a aquisição de tokens e a montagem dos cabeçalhos em funções auxiliares — em vez de adquirir tokens em cada ponto de chamada — oferece três vantagens concretas:

• Credencial centralizada: um único DefaultAzureCredential é encapsulado em TokenProvider e reutilizado para cada chamada de API, de modo que a descoberta de identidade (CLI do Azure, VS Code, identidade gerenciada etc.) aconteça uma vez.
• Tokens com reconhecimento de público: os pontos de extremidade de cluster do Fabric, do Kusto e do Power BI rejeitam tokens emitidos para o público incorreto. Um gerador de cabeçalhos separado para cada público mantém o escopo correto ao lado do local da chamada, o que deixa claro a qual endpoint cada função se destina.
• Renovado a cada solicitação: os construtores de cabeçalho constroem o cabeçalho Authorization sob demanda, em vez de armazená-lo em cache. A credencial subjacente é atualizada de forma transparente, portanto, os sites de chamadas nunca precisam pensar em expiração.

Este tutorial usa as APIs REST do Fabric com escopos delegados, como Item.ReadWrite.All.

Adicione o seguinte após a Config classe:

# =========================================================
# Auth helpers
#
# Authentication utilities built on `DefaultAzureCredential` that acquire and
# construct Authorization headers for calling Fabric REST APIs.
# =========================================================

class TokenProvider:
    """
    Thin wrapper around `DefaultAzureCredential` that acquires Entra access
    tokens. `_fabric_headers()` and `_pbi_headers()` call `get()` per
    request so the Authorization header is always fresh; the underlying
    credential refreshes transparently.
    """
    def __init__(self):
        self._cred = DefaultAzureCredential()

    def get(self, scope: str) -> str:
        return self._cred.get_token(scope).token


_tokens = TokenProvider()


def _fabric_headers() -> dict[str, str]:
    """
    Build headers for Fabric REST API calls.

    This function is called each time we make a Fabric REST call so the token is fresh.
    """
    return {
        "Authorization": f"Bearer {_tokens.get('https://api.fabric.microsoft.com/.default')}",
        "Content-Type": "application/json"
    }


def _kusto_headers() -> dict[str, str]:
    """
    Build headers for Kusto (Eventhouse `queryServiceUri`) management and query calls.
    """
    return {
        "Authorization": f"Bearer {_tokens.get('https://api.kusto.windows.net/.default')}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }


def _pbi_headers() -> dict[str, str]:
    """
    Build headers for polling Power BI cluster LRO endpoints
    (e.g., df-*.analysis.windows.net) that require a Power BI audience token.
    """
    return {
        "Authorization": f"Bearer {_tokens.get('https://analysis.windows.net/powerbi/api/.default')}",
        "Content-Type": "application/json"
    }

Criar um wrapper de cliente Fabric

A maioria das chamadas REST Fabric neste tutorial envia os mesmos cabeçalhos Authorization e Content-Type. Em vez de repeti-los em cada local de chamada, este tutorial encapsula httpx.Client em um pequeno FabricClient que adiciona os cabeçalhos automaticamente, mas ainda retorna o httpx.Response bruto para que cada chamada possa inspecionar os códigos de status (por exemplo, para distinguir 201 de 202).

Encapsular httpx.Client dessa forma — em vez de passar headers=_fabric_headers() em todas as chamadas — oferece duas vantagens concretas:

• Cabeçalhos em um só lugar: Cada local de chamada usa automaticamente o _fabric_headers() mais recente, assim uma nova solicitação não pode ser enviada acidentalmente sem o cabeçalho Authorization.
• Os códigos de status permanecem visíveis: request() retorna o bruto httpx.Response em vez de JSON decodificado, para que os sites de chamadas ainda possam ramificar o status (201 vs 202) e inspecionar cabeçalhos como Location ou Retry-After para manipulação de LRO.

Adicione o seguinte após as funções auxiliares de autenticação:

# =========================================================
# FabricClient (minimal wrapper so call sites stay clean)
# =========================================================

class FabricClient:
    """
    Small wrapper around httpx.Client so we don't repeat headers everywhere.

    Keeps the tutorial behavior:
    - request() returns the raw httpx.Response so the caller can handle 201 vs 202.
    """
    def __init__(self, http_client: httpx.Client):
        self._http = http_client

    def request(self, method: str, url: str, *, json_body=None) -> httpx.Response:
        return self._http.request(method, url, headers=_fabric_headers(), json=json_body)

Criar uma função auxiliar de LRO

As várias APIs REST do Fabric usadas neste tutorial — como Criar Casa de Eventos, Criar Fluxo de Eventos e Criar Mapa — dão suporte a operações de longa duração (LROs).

Essas APIs podem retornar respostas em vários padrões:

• 201 Created com o corpo do recurso em linha (síncrono)
• 202 Accepted com um Location cabeçalho que aponta para uma URL de status da operação (assíncrona)
• 202 Accepted com um x-ms-operation-id cabeçalho em vez de Location (formulário assíncrono e alternativo)
• 200 OK com status: "Running" ou status: "NotStarted" durante a sondagem (ainda em andamento)
• 200 OK com status: "Succeeded", mas nenhuma ID do recurso no corpo (bem-sucedido; resolver listando e fazendo a correspondência de displayName)

Para lidar com tudo isso de forma consistente, você cria uma única função auxiliar que:

1. Retorna a ID do recurso imediatamente se a resposta inicial já a contiver.
2. Caso contrário, consulta a URL da operação (criada a partir de Location ou x-ms-operation-id) usando Retry-After.
3. 200 OK Trata como status: "Running" / "NotStarted" ainda em andamento e continua a sondagem.
4. Em caso de sucesso, retorna o ID do recurso no corpo da resposta ou, caso o corpo contenha apenas o status, recorre à listagem de recursos e faz a correspondência com base em displayName (com tentativas repetidas).
5. Usa _pbi_headers() quando a URL de consulta estiver em um cluster do Power BI (*.analysis.windows.net) e os cabeçalhos do Fabric, caso contrário.

Esta única função auxiliar substitui a necessidade de funções auxiliares de "resolver por nome" para cada recurso — todas as funções create_* neste tutorial chamam _handle_lro com o list_url e o match_display_name apropriados.

Adicione o seguinte após a FabricClient classe:

# =========================================================
# LRO handler 
# =========================================================

def _handle_lro(
    client: httpx.Client,
    initial_response: httpx.Response,
    *,
    list_url: str | None = None,
    match_display_name: str | None = None,
    id_field: str = "id",
    max_attempts: int = 10,
    delay: int = 5,
) -> str:
    """
    Handle a Fabric long-running operation (LRO) and return the resource id.

    Supports the response patterns used by Fabric REST APIs:
    - 200/201 with the resource body inline (synchronous).
    - 202 with a `Location` header or `x-ms-operation-id` (asynchronous).
    - 200 with `status: "Running"` / `"NotStarted"` while polling.
    - 200 with `status: "Succeeded"` but no id (resolve by listing and matching `displayName`).

    Polling uses `Retry-After` and switches to a Power BI audience token when
    the operation URL is on `*.analysis.windows.net`.
    """
    # Sync 200/201 with body: return the id immediately.
    if initial_response.status_code in (200, 201):
        try:
            body = initial_response.json() if initial_response.content else {}
        except ValueError:
            body = {}
        if isinstance(body, dict) and body.get(id_field):
            return body[id_field]

    # Location header, with x-ms-operation-id fallback.
    op_url = initial_response.headers.get("Location")
    if not op_url:
        op_id = initial_response.headers.get("x-ms-operation-id")
        if op_id:
            op_url = f"https://api.fabric.microsoft.com/v1/operations/{op_id}"
        else:
            raise RuntimeError(
                f"Missing LRO Location/x-ms-operation-id. "
                f"status={initial_response.status_code} body={initial_response.text[:500]!r}"
            )

    # Audience-aware polling: Power BI cluster endpoints need a different token.
    poll_headers = _pbi_headers() if "analysis.windows.net" in op_url else _fabric_headers()
    retry_after = int(initial_response.headers.get("Retry-After", "5"))

    while True:
        time.sleep(retry_after)
        poll = client.get(op_url, headers=poll_headers)

        if poll.status_code == 202:
            retry_after = int(poll.headers.get("Retry-After", "5"))
            continue

        poll.raise_for_status()
        body = poll.json() if poll.content else {}
        status = body.get("status") if isinstance(body, dict) else None

        if status in ("Running", "NotStarted"):
            retry_after = int(poll.headers.get("Retry-After", "5"))
            continue
        if status == "Failed":
            raise RuntimeError(f"LRO failed. Body: {body}")

        if isinstance(body, dict) and body.get(id_field):
            return body[id_field]

        # Status-only success: list and match by displayName, with retries.
        if status == "Succeeded" and list_url and match_display_name:
            for attempt in range(max_attempts):
                r = client.get(list_url, headers=_fabric_headers())
                r.raise_for_status()
                match = next(
                    (i for i in r.json().get("value", []) if i.get("displayName") == match_display_name),
                    None,
                )
                if match and match.get(id_field):
                    return match[id_field]
                time.sleep(delay)
            raise RuntimeError(
                f"LRO succeeded but resource not visible after retries. "
                f"match_display_name={match_display_name!r}"
            )

        raise RuntimeError(f"LRO completed but no resource id was returned. Body: {body}")

Auxiliar de carga útil de definição

Quando você cria um mapa com uma definição pública, a API REST de criação de mapas espera que cada parte em definition.parts contenha uma carga útil codificada em base64 com "payloadType": "InlineBase64". O auxiliar _json_to_b64 codifica um dict do Python (o seu map.json) para esse formato, para que create_map possa inseri-lo diretamente no corpo da solicitação.

Adicione o seguinte após a _handle_lro função:

# =========================================================
# Definition payload helper
#
# Encodes map.json as base64 for inline Create map payloads.
# =========================================================

def _json_to_b64(obj: dict) -> str:
    """
    Convert a Python dict to base64-encoded JSON text.

    Fabric Map "Create Map with definition inline" requires:
    - definition.parts[].payloadType = InlineBase64
    - definition.parts[].payload     = base64(json(map_json))
    """
    return base64.b64encode(json.dumps(obj).encode("utf-8")).decode("utf-8")

Criar utilitário para fornecer a cadeia de conexão do fluxo de eventos

Para enviar eventos para o ponto de extremidade personalizado do fluxo de eventos, o script precisa de um cadeia de conexão para esse ponto de extremidade.

Ao contrário das APIs REST do Fabric que você vem chamando até agora (que são operações de plano de controle para criar e gerenciar recursos), a ingestão do fluxo de eventos usa um ponto de extremidade de plano de dados compatível com o Event Hubs, e esse ponto de extremidade é autenticado com uma cadeia de conexão baseada em SAS, em vez de um token do Microsoft Entra. A cadeia de conexão é criada quando você adiciona a origem de ponto de extremidade personalizada e não é exposta pela API REST do Fabric, portanto, ela precisa ser copiada do portal do Fabric.

get_eventhub_connection_string_interactive reutiliza um valor da variável de ambiente EVENTHUB_CONNECTION_STRING (útil em execuções repetidas) ou solicita esse valor durante a execução e, em seguida, o armazena em cache em cfg para que as etapas posteriores possam reutilizá-lo sem solicitá-lo novamente.

Adicione o seguinte após a _json_to_b64 função:

def get_eventhub_connection_string_interactive(cfg: Config) -> str:
    """
    Prompt for (or read) the eventstream custom endpoint connection string.

    The connection string is created when the custom endpoint source is added
    to the eventstream and isn't exposed by the Fabric REST API, so we read it
    from the `EVENTHUB_CONNECTION_STRING` environment variable when set, or
    prompt interactively otherwise. The value is cached on `cfg` for reuse.
    """
    if getattr(cfg, "eventhub_connection_string", None):
        return cfg.eventhub_connection_string

    print("\n=== Eventstream connection string required ===")
    print("In the Fabric portal:")
    print("  1) Open the eventstream you just created")
    print("  2) Select the custom endpoint source")
    print("  3) Select SAS Key Authentication")
    print("  4) Copy Connection string-primary key\n")

    cfg.eventhub_connection_string = input("Paste connection string here: ").strip()

    if not cfg.eventhub_connection_string:
        raise RuntimeError("Connection string cannot be empty.")

    return cfg.eventhub_connection_string

Criar auxiliar para propagar eventos iniciais do CSV

Para garantir que o mapa exiba dados imediatamente após sua criação, o script envia um pequeno conjunto de eventos de semente para o fluxo de eventos antes de criar o mapa.

Sem essa etapa, a tabela Eventhouse ainda pode não conter dados e o mapa pode aparecer vazio na primeira carga.

Essa função auxiliar lê dados de um arquivo CSV local e envia cada linha como um evento JSON para o fluxo de eventos usando o protocolo EventHub.

Como os recursos do Eventstream são provisionados de forma assíncrona, o endpoint personalizado pode não estar imediatamente pronto para aceitar eventos após sua criação. Para lidar com isso, a função auxiliar inclui um mecanismo interno de repetição que tenta automaticamente enviar eventos até que o endpoint esteja disponível. Isso garante que o processo de inicialização seja confiável e repetível e não exija ajustes manuais de tempo.

Essa abordagem espelha padrões de ingestão do mundo real:

• Os dados são produzidos externamente (por exemplo, dispositivos IoT ou aplicativos)
• Os eventos são transmitidos para Eventstream
• O Eventstream fornece dados ao Eventhouse para consulta e visualização

Ao inserir eventos iniciais, você simula esse fluxo de ingestão e garante que:

• A tabela de destino é preenchida
• A função KQL tem dados a serem retornados
• O mapa é renderizado imediatamente após a criação

Adicione o seguinte código após a get_eventhub_connection_string_interactive() função:

def seed_eventstream_from_csv(cfg: Config, max_attempts: int = 10, delay: int = 3) -> int:
    """
    Send seed events from a CSV with retries to handle eventstream readiness delay.
    """
    conn_str = get_eventhub_connection_string_interactive(cfg)

    last_error = None
    for attempt in range(1, max_attempts + 1):
        print(f"Seeding attempt {attempt}/{max_attempts}...")
        try:
            sent = 0
            producer = EventHubProducerClient.from_connection_string(conn_str=conn_str)
            try:
                with open(cfg.seed_csv_path, newline="", encoding="utf-8") as f:
                    reader = csv.DictReader(f)
                    batch = producer.create_batch()
                    batch_count = 0
                    for row in reader:
                        event = {
                            "VehicleId": row["VehicleId"],
                            "Latitude": float(row["Latitude"]),
                            "Longitude": float(row["Longitude"]),
                            "EventTime": row["EventTime"],
                        }
                        data = EventData(json.dumps(event))
                        try:
                            batch.add(data)
                            batch_count += 1
                        except ValueError:
                            producer.send_batch(batch)
                            sent += batch_count
                            batch = producer.create_batch()
                            batch.add(data)
                            batch_count = 1
                    if batch_count > 0:
                        producer.send_batch(batch)
                        sent += batch_count
                print(f"Seed events sent: {sent}")
                return sent
            finally:
                producer.close()
        except (EventHubError, ConnectionError, TimeoutError) as exc:
            last_error = exc
            print(f"Seeding failed (attempt {attempt}): {exc}. Retrying in {delay}s...")
            time.sleep(delay)

    raise RuntimeError(f"Seeding failed after {max_attempts} attempts. Last error: {last_error}")

Aguardar a disponibilidade do banco de dados KQL

Depois que o eventhouse, seu banco de dados KQL e a função KQL existirem, o banco de dados KQL ainda pode não ser localizado imediatamente por outros endpoints REST do Fabric. Os serviços do Fabric são executados em back-ends distribuídos, por isso um recurso recém-criado pode levar algum tempo para se propagar entre eles.

Se você chamar Criar Mapa imediatamente após a função KQL estar em vigor, criar mapa poderá falhar ao resolver a fonte de dados e retornar um erro como o banco de dados Kusto não encontrado.

wait_for_kql_database_ready consulta o endpoint REST do Fabric para o banco de dados KQL e retorna assim que há uma resposta 200 OK. É uma verificação de melhor esforço — uma resposta bem-sucedida nesse endpoint do plano de controle é um forte indicativo de que o Maps também consegue resolver o banco de dados — e gera RuntimeError após max_attempts se o banco de dados nunca ficar visível.

Adicione o seguinte após a seed_eventstream_from_csv função:

# =========================================================
# KQL database readiness helper
#
# Polls the KQL database's Fabric REST endpoint until it
# responds 200, as a best-effort gate before Create Map
# references it as a data source.
# =========================================================

def wait_for_kql_database_ready(
    client: httpx.Client,
    cfg: Config,
    kql_database_item_id: str,
    max_attempts: int = 10,
    delay: int = 3,
) -> None:
    """
    Poll the Fabric REST endpoint for a KQL database until it returns 200.

    Acts as a best-effort readiness gate before calling Create Map with the
    KQL database as a data source. Retries `max_attempts` times with `delay`
    seconds between attempts, then raises `RuntimeError` if the database
    never becomes visible.
    """
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    for attempt in range(1, max_attempts + 1):
        resp = client.get(url, headers=_fabric_headers())
        if resp.status_code == 200:
            print("KQL database is available to Fabric Maps")
            return
        print(
            f"Waiting for KQL database availability "
            f"(attempt {attempt}/{max_attempts}, status={resp.status_code})..."
        )
        time.sleep(delay)

    raise RuntimeError(
        f"KQL database {kql_database_item_id!r} did not become available after {max_attempts} attempts."
    )

Criar funções primárias

Em seguida, adicione as funções primárias que definem o fluxo de trabalho. Todos eles são chamados a partir de main().

As funções são adicionadas na ordem em que são definidas no código. main() chama-as em uma ordem ligeiramente diferente para que a tabela KQL exista antes que o fluxo de eventos se associe a ela e, portanto, os dados semeados estão disponíveis antes da execução da verificação.

• Criar uma casa de eventos
• Criar a tabela KQL
• Verificar ingestão (chamada após a semeadura)
• Criar um fluxo de eventos
• Criar uma função KQL
• Criar a definição do mapa (map.json)
• Compilar os metadados da plataforma (.platform)
• Criar o mapa

Criar uma casa de eventos

create_eventhouse cria um eventhouse no espaço de trabalho e retorna o ID do item. A API REST Create Eventhouse pode atender à mesma chamada de três maneiras diferentes:

• 201 Created com o ID do eventhouse embutido em linha (de forma síncrona).
• 202 Accepted com uma URL de operação LRO (assíncrona).
• 409 Conflict com x-ms-public-api-error-code definido para ItemDisplayNameNotAvailableYet (o nome anterior ainda está reservado no backend) ou ItemDisplayNameAlreadyInUse (já existe um eventhouse com esse nome no espaço de trabalho).

Para lidar com os três de forma confiável: create_eventhouse

• Encaminha as respostas de 201 e 202 para _handle_lro, que já lida com a conclusão síncrona e assíncrona de forma uniforme.
• Respeita Retry-After e tenta novamente (até cinco tentativas) em ItemDisplayNameNotAvailableYet.
• Reutiliza o eventhouse existente em ItemDisplayNameAlreadyInUse listando os eventhouses no espaço de trabalho e fazendo a correspondência por displayName.
• Recorre a um nome de exibição exclusivo (com o sufixo de um UUID curto) se o nome nunca ficar disponível após o limite de tentativas ser esgotado, para que o script ainda possa prosseguir.

Adicione o seguinte após a wait_for_kql_database_ready função:

# =========================================================
# Create an eventhouse
# =========================================================

def create_eventhouse(client: httpx.Client, fabric: FabricClient, cfg: Config) -> str:
    """
    Create an eventhouse in the workspace and return its item ID.

    Handles the three response patterns Create Eventhouse can return:
    - 201/202: delegate to `_handle_lro` (synchronous body or LRO completion).
    - 409 `ItemDisplayNameNotAvailableYet`: honor `Retry-After` and retry.
    - 409 `ItemDisplayNameAlreadyInUse`: list eventhouses and reuse the one
      whose `displayName` matches `cfg.eventhouse_display_name`.

    If the name remains unavailable after the retry budget, falls back to a
    uniquified display name (suffixed with a short UUID) so the script can
    still make forward progress.
    """
    eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses"
    eventhouse_payload = {
        "displayName": cfg.eventhouse_display_name,
        "description": cfg.eventhouse_description
    }

    # Retry loop to handle transient "name not available yet"
    for attempt in range(1, 6):
        eh_resp = fabric.request("POST", eventhouse_url, json_body=eventhouse_payload)

        print("Create eventhouse status:", eh_resp.status_code)
        print("Create eventhouse headers:", dict(eh_resp.headers))

        # 201/202: success or LRO — _handle_lro handles both.
        if eh_resp.status_code in (201, 202):
            eventhouse_id = _handle_lro(
                client, eh_resp,
                list_url=eventhouse_url,
                match_display_name=cfg.eventhouse_display_name,
            )
            print("Eventhouse created. Eventhouse ID:", eventhouse_id)
            return eventhouse_id

        # 409: name issues
        if eh_resp.status_code == 409:
            api_code = eh_resp.headers.get("x-ms-public-api-error-code")

            # Name reserved temporarily: wait and retry
            if api_code == "ItemDisplayNameNotAvailableYet":
                wait_s = int(eh_resp.headers.get("retry-after", "20"))
                print(f"Name not available yet (attempt {attempt}/5). Waiting {wait_s}s then retrying...")
                time.sleep(wait_s)
                continue

            # Name already exists: reuse existing eventhouse by displayName
            if api_code == "ItemDisplayNameAlreadyInUse":
                print(f"Eventhouse {cfg.eventhouse_display_name!r} already exists. Reusing it...")
                r = client.get(eventhouse_url, headers=_fabric_headers())
                r.raise_for_status()
                match = next(
                    (i for i in r.json().get("value", []) if i.get("displayName") == cfg.eventhouse_display_name),
                    None,
                )
                if match and match.get("id"):
                    return match["id"]
                raise RuntimeError(
                    f"Eventhouse {cfg.eventhouse_display_name!r} reported as existing but not found in list."
                )

        # Anything else: fail fast with details
        raise RuntimeError(f"Create eventhouse failed: {eh_resp.status_code} {eh_resp.text}")

    # If the name never becomes available, last-resort: pick a unique name and try once
    cfg.eventhouse_display_name = f"{cfg.eventhouse_display_name}-{uuid.uuid4().hex[:8]}"
    print(f"Name still not available; switching to unique name: {cfg.eventhouse_display_name}")
    eh_resp = fabric.request("POST", eventhouse_url, json_body={
        "displayName": cfg.eventhouse_display_name,
        "description": cfg.eventhouse_description
    })
    eh_resp.raise_for_status()
    return eh_resp.json()["id"]

Criar tabela KQL

create_kql_table_if_missing garante que a tabela de destino exista no banco de dados KQL antes que o fluxo de eventos comece a gravar nela. O destino eventstream criado posteriormente é configurado com ProcessedIngestion e um fixo tableName, portanto, a tabela já deve existir quando os eventos chegam – caso contrário, a ingestão falhará.

A função executa um comando .create-merge table no endpoint de gerenciamento do Kusto do eventhouse (queryServiceUri + /v1/rest/mgmt). .create-merge é idempotente: ele cria a tabela se ela não existir e mescla o esquema se ela existir. Isso torna seguro chamar em cada execução.

Antes de emitir o comando, a função lê as propriedades do Eventhouse para obter queryServiceUri e o ID do item do banco de dados KQL e, em seguida, resolve o displayName do banco de dados para que a carga útil mgmt faça referência a ele pelo nome, e não pelo ID.

Adicione o seguinte após a create_eventhouse função:

# =========================================================
# Create the KQL table (idempotent)
# =========================================================

def create_kql_table_if_missing(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str) -> None:
    """
    Create or merge the destination table in the eventhouse's KQL database.

    Reads the eventhouse properties to discover `queryServiceUri` and the KQL
    database item ID, resolves the database's `displayName`, then issues a
    `.create-merge table` command against the Kusto management endpoint.
    `.create-merge` is idempotent: it creates the table if missing and merges
    the schema if it already exists.
    """
    # Get queryServiceUri + KQL database item id
    get_eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}"
    eh = fabric.request("GET", get_eventhouse_url)
    eh.raise_for_status()

    props = eh.json().get("properties") or {}
    query_service_uri = props.get("queryServiceUri")
    databases_item_ids = props.get("databasesItemIds") or []
    if not query_service_uri or not databases_item_ids:
        raise RuntimeError("Eventhouse missing queryServiceUri or databasesItemIds")

    kql_database_item_id = databases_item_ids[0]

    # Resolve actual DB displayName (don't rely on cfg.kql_database_name)
    get_db_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    db_resp = fabric.request("GET", get_db_url)
    db_resp.raise_for_status()
    kql_database_name = db_resp.json().get("displayName")
    if not kql_database_name:
        raise RuntimeError("KQL database response did not include displayName")

    # Create (or merge) the table schema
    # (Schema matches what your CSV sends: VehicleId, Latitude, Longitude, EventTime)
    csl = f""".create-merge table {cfg.eventhouse_table_name} (
        VehicleId: string,
        Latitude: real,
        Longitude: real,
        EventTime: datetime
    )"""

    mgmt_url = f"{query_service_uri}/v1/rest/mgmt"
    mgmt_payload = {"db": kql_database_name, "csl": csl}
    resp = client.post(mgmt_url, headers=_kusto_headers(), json=mgmt_payload)
    if resp.status_code >= 400:
        raise RuntimeError(f"Create table failed: {resp.status_code}\n{resp.text}")

    print(f"Ensured table exists: {cfg.eventhouse_table_name}")

Verificar ingestão de dados

verify_eventhouse_data confirma que os eventos inseridos no Eventstream foram realmente gravados na tabela Eventhouse. Ele executa consultas periódicas de uma <table> | count consulta no endpoint de consulta do Kusto do eventhouse (queryServiceUri + /v1/rest/query) até que a contagem retornada seja maior que zero ou falha após expirar o tempo limite. A ingestão de fluxo de eventos leva alguns segundos para fluir do ponto de extremidade personalizado para a tabela, portanto, sondar — em vez de uma única consulta — é o que fornece um sinal de aprovação/falha confiante.

Ele é definido ao lado de create_kql_table_if_missing porque ambas as funções auxiliares consultam as mesmas propriedades do eventhouse (queryServiceUri, databasesItemIds) e resolvem o displayName do banco de dados KQL. É chamado a partir de main()afterseed_eventstream_from_csv para que os eventos iniciais tenham a chance de fluir pelo stream de eventos e alcançar a tabela antes de a contagem ser executada.

Executar essa verificação logo no início identifica cedo uma configuração incorreta de ingestão — por exemplo, um destino de fluxo de eventos apontando para o nome de tabela errado — em vez de deixar que isso só apareça mais tarde na forma de um mapa vazio.

Adicione o seguinte após a create_kql_table_if_missing função:

# =========================================================
# Verify data ingestion (called after seeding)
# =========================================================

def verify_eventhouse_data(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str):
    """
    Poll a count query against the eventhouse table until rows arrive.

    Reads the eventhouse properties to get `queryServiceUri` and the KQL
    database item ID, resolves the database's `displayName`, then polls
    `<table> | count` against the Kusto query endpoint
    (`queryServiceUri` + `/v1/rest/query`) until the count is greater than
    zero or the timeout elapses. Eventstream ingestion is asynchronous, so
    polling avoids a false negative when the query runs before seeded
    events have landed in the table.
    """

    # Reuse your existing pattern to get KQL DB info
    eh = fabric.request("GET", f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}")
    eh.raise_for_status()

    props = eh.json().get("properties") or {}
    db_ids = props.get("databasesItemIds") or []
    query_service_uri = props.get("queryServiceUri")

    if not db_ids or not query_service_uri:
        raise RuntimeError("Missing eventhouse properties for verification")

    db_id = db_ids[0]

    db_resp = fabric.request("GET", f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{db_id}")
    db_resp.raise_for_status()

    db_name = db_resp.json().get("displayName")

    # Simple count query
    csl = f"{cfg.eventhouse_table_name} | count"
    query_url = f"{query_service_uri}/v1/rest/query"

    max_attempts = 12
    delay_seconds = 5

    for attempt in range(1, max_attempts + 1):
        resp = client.post(
            query_url,
            headers=_kusto_headers(),
            json={"db": db_name, "csl": csl}
        )

        if resp.status_code >= 400:
            raise RuntimeError(f"Verification query failed: {resp.text}")

        # Kusto v1 query response: Tables[0].Rows[0][0] holds the count.
        count = resp.json()["Tables"][0]["Rows"][0][0]
        print(f"Data verification attempt {attempt}/{max_attempts}: count = {count}")

        if count > 0:
            print(f"Data ingestion verified: {count} row(s) in {cfg.eventhouse_table_name}")
            return

        if attempt < max_attempts:
            time.sleep(delay_seconds)

    raise RuntimeError(
        f"Data verification failed: no rows in {cfg.eventhouse_table_name} after "
        f"{max_attempts * delay_seconds}s"
    )

Criar eventstream com definição

create_eventstream_with_definition cria um fluxo de eventos no workspace com sua topologia completa inserida na solicitação e retorna a ID do item do fluxo de eventos. O uso de uma definição pública permite provisionar o fluxo de eventos e conectar suas fontes, fluxos e destinos em uma única chamada, em vez de criar o fluxo de eventos primeiro e, em seguida, aplicar patch em sua definição.

Antes de enviar a solicitação, a função lê as propriedades do Eventhouse para obter o ID do item do banco de dados KQL e resolve o displayName do banco de dados para que o destino faça referência a ele pelo nome, e não pelo ID. Em seguida, ele cria um grafo de fluxo de eventos com uma origem CustomEndpoint, um DefaultStream e um destino Eventhouse configurado com ProcessedIngestion e o tableName fixo de cfg, codifica o grafo em base64 como a parte eventstream.json e o envia por POST para Create Eventstream.

A API REST Create Eventstream pode retornar 201 Created (síncrono, corpo em linha), 202 Accepted (LRO assíncrona por meio de Location ou x-ms-operation-id) ou 200 OK com um payload de conclusão apenas com status, em que o fluxo de eventos ainda não está visível na resposta de List Eventstreams devido ao atraso na propagação do back-end. _handle_lro abrange todos esses casos , incluindo listagem e correspondência por displayName – de modo que essa função delega o tratamento de resposta completo a ele em uma única chamada.

Adicione o seguinte após a verify_eventhouse_data função:

# =========================================================
# Create eventstream with definition
# =========================================================

def create_eventstream_with_definition(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str) -> str:
    """
    Create an eventstream with a public definition and return its item ID.

    Reads the eventhouse properties to discover the KQL database item ID and
    resolves the database's `displayName`, then builds an eventstream graph
    with a `CustomEndpoint` source, a `DefaultStream`, and an `Eventhouse`
    destination configured with `ProcessedIngestion` and the table name from
    `cfg`. Base64-encodes the graph as the `eventstream.json` part, POSTs it
    to Create Eventstream, and delegates response handling to `_handle_lro`.
    """
    eventstream_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventstreams"

    source_name = "CustomEndpointSource"
    stream_name = "DefaultStream"
    destination_name = "EventhouseDestination"

    # Resolve the KQL database item ID from the Eventhouse
    get_eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}"
    eh = fabric.request("GET", get_eventhouse_url)
    eh.raise_for_status()

    props = (eh.json().get("properties") or {})
    databases_item_ids = props.get("databasesItemIds") or []
    if not databases_item_ids:
        raise RuntimeError("Eventhouse properties did not include databasesItemIds.")

    kql_database_item_id = databases_item_ids[0]

    # Resolve the actual KQL database *name* (displayName) to avoid name drift
    get_db_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    db_resp = fabric.request("GET", get_db_url)
    db_resp.raise_for_status()
    kql_database_name = db_resp.json().get("displayName")
    if not kql_database_name:
        raise RuntimeError("KQL database response did not include displayName.")

    print("Eventhouse ID:", eventhouse_id)
    print("KQL database item ID:", kql_database_item_id)
    print("KQL database name:", kql_database_name)

    eventstream_json = {
        "sources": [
            {
                "name": source_name,
                "type": "CustomEndpoint",
                "properties": {
                    "inputSerialization": {"type": "Json", "properties": {"encoding": "UTF8"}}
                }
            }
        ],
        "streams": [
            {
                "name": stream_name,
                "type": "DefaultStream",
                "properties": {},
                "inputNodes": [{"name": source_name}]
            }
        ],
        "operators": [],
        "destinations": [
            {
                "name": destination_name,
                "type": "Eventhouse",
                "properties": {
                    "dataIngestionMode": "ProcessedIngestion",
                    "workspaceId": cfg.workspace_id,
                    "itemId": kql_database_item_id,
                    "databaseName": kql_database_name,
                    "tableName": cfg.eventhouse_table_name,
                    "inputSerialization": {"type": "Json", "properties": {"encoding": "UTF8"}}
                },
                "inputNodes": [{"name": stream_name}]
            }
        ],
        "compatibilityLevel": "1.1"
    }

    eventstream_payload = {
        "displayName": cfg.eventstream_display_name,
        "description": cfg.eventstream_description,
        "definition": {
            "parts": [
                {
                    "path": "eventstream.json",
                    "payload": _json_to_b64(eventstream_json),
                    "payloadType": "InlineBase64"
                }
            ]
        }
    }

    es_resp = fabric.request("POST", eventstream_url, json_body=eventstream_payload)

    eventstream_id = _handle_lro(
        client,
        es_resp,
        list_url=eventstream_url,
        match_display_name=cfg.eventstream_display_name,
    )

    print("Eventstream created. Eventstream ID:", eventstream_id)
    return eventstream_id

Criar função KQL

create_kql_function cria (ou atualiza) uma função Kusto armazenada no banco de dados KQL do eventhouse e retorna a ID do item do banco de dados KQL para que o chamador possa transferir a fonte de dados do mapa para ele. A função — LatestVehicleLocations, por padrão — retorna a linha mais recente de cada VehicleId via arg_max(EventTime, *), projetando Latitude, Longitude, VehicleId e EventTime para que o Fabric Maps possa vincular as colunas de latitude e longitude da camada.

Assim como create_kql_table_if_missing, este auxiliar é executado no endpoint de gerenciamento Kusto do eventhouse (queryServiceUri + /v1/rest/mgmt) e é idempotente: .create-or-alter function cria a função se não existir e substitui o corpo dela se já existir, portanto, é seguro chamá-lo em todas as execuções.

O comando é enviado com skipvalidation=true porque o corpo da função faz referência à tabela de destino por meio de table("<name>"), em vez de como um identificador simples. A forma table() adia a resolução de nomes para o momento da consulta; caso contrário, a validação na criação falharia se a tabela ainda não tivesse recebido nenhum dado e seu schema não estivesse totalmente visível para o validador. O emparelhamento skipvalidation=true com table("...") permite que a função seja criada antes que a ingestão preencha a tabela, que é a ordem em que este tutorial é executado.

Adicione o seguinte após a create_eventstream_with_definition função:

# =========================================================
# Create KQL function
# =========================================================

def create_kql_function(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str) -> str:
    """
    Create or update the stored Kusto function used by the map layer.

    Reads the eventhouse properties to discover `queryServiceUri` and the
    KQL database item ID, resolves the database's `displayName`, then
    issues a `.create-or-alter function` command against the Kusto
    management endpoint with `skipvalidation=true` and a `table("...")`
    reference so the function can be created before the destination table
    has any data. Returns the KQL database item ID so the caller can wire
    the map's data source to it.
    """
    # Get eventhouse properties (queryServiceUri + databasesItemIds)
    get_eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}"
    eh = fabric.request("GET", get_eventhouse_url)
    eh.raise_for_status()

    props = (eh.json().get("properties") or {})
    query_service_uri = props.get("queryServiceUri")
    databases_item_ids = props.get("databasesItemIds") or []

    if not query_service_uri:
        raise RuntimeError("Eventhouse properties did not include queryServiceUri.")
    if not databases_item_ids:
        raise RuntimeError("Eventhouse properties did not include databasesItemIds.")

    # We'll return this so the caller can wire the map to the correct KQL database item id.
    kql_database_item_id = databases_item_ids[0]

    # Resolve actual KQL database name (displayName)
    get_db_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    db_resp = fabric.request("GET", get_db_url)
    db_resp.raise_for_status()

    kql_database_name = db_resp.json().get("displayName")
    if not kql_database_name:
        raise RuntimeError("KQL database response did not include displayName.")

    # Create the function that returns the latest location per vehicle.
    # Keep columns explicit so the map config can bind Latitude/Longitude.
    kql = f""".create-or-alter function with (skipvalidation=true) {cfg.kql_function_name}() {{
    table("{cfg.eventhouse_table_name}")
    | summarize arg_max(EventTime, *) by VehicleId
    | project Latitude, Longitude, VehicleId, EventTime
    }}"""

    mgmt_url = f"{query_service_uri}/v1/rest/mgmt"
    mgmt_payload = {"db": kql_database_name, "csl": kql}

    mgmt_resp = client.post(mgmt_url, headers=_kusto_headers(), json=mgmt_payload)

    if mgmt_resp.status_code >= 400:
        # Kusto usually returns a detailed JSON error body on 400s.
        raise RuntimeError(
            "Kusto mgmt call failed.\n"
            f"URL: {mgmt_url}\n"
            f"DB: {mgmt_payload.get('db')}\n"
            f"Status: {mgmt_resp.status_code}\n"
            f"Body: {mgmt_resp.text}"
        )

    print("KQL function created/updated:", cfg.kql_function_name)
    return kql_database_item_id

Gerar map.json

build_map_json compila e retorna a carga de map.json que define o conteúdo do mapa de Fabric. O conteúdo segue o esquema de definição de item do mapa e é composto por quatro seções: dataSources (de onde os dados vêm), iconSources (marcadores personalizados opcionais), layerSources (o que é consultado e com que frequência) e layerSettings (como o resultado é renderizado no mapa).

Para este tutorial, dataSources aponta para o banco de dados KQL (itemType: "KqlDatabase") criado anteriormente, e a única entrada em layerSources é uma camada com suporte do Kusto (type: "kusto", queryType: "function"), cujo query chama a função armazenada LatestVehicleLocations(). refreshIntervalMs é lido a partir de cfg.refresh_interval_ms (5000 ms por padrão), de modo que a camada reexecuta a função em intervalos regulares, e o mapa reflete os novos dados ingeridos em tempo quase real.

A entrada correspondente layerSettings associa as colunas de resultado da camada ao mapa por meio de latitudeColumnName: "Latitude" e longitudeColumnName: "Longitude", renderiza cada linha como um ponto bubble e exibe VehicleId e EventTime em dicas de ferramenta. A função imprime o conteúdo montado para que você possa inspecionar o JSON exato que a chamada Criar Mapa envia.

Para obter mais informações sobre a API REST de definição de mapa, consulte a definição de item do mapa.

Adicione o seguinte após a create_kql_function função:

# =========================================================
# Build map.json
# =========================================================

def build_map_json(cfg: Config, kql_database_item_id: str) -> dict:
    """
    Build and return the map.json payload for the Fabric Map.

    Wires `dataSources` to the KQL database created earlier, defines a
    single Kusto-backed layer in `layerSources` that calls the stored
    function `cfg.kql_function_name` and re-runs it every
    `cfg.refresh_interval_ms` milliseconds, and configures `layerSettings`
    to bind the `Latitude` / `Longitude` columns and render each row as a
    bubble point. Prints the assembled payload for inspection.
    """
    layer_source_id = str(uuid.uuid4())
    layer_setting_id = str(uuid.uuid4())
    data_source_name = "kqlConnection"

    map_json = {
        "$schema": "https://developer.microsoft.com/json-schemas/fabric/item/map/definition/2.0.0/schema.json",
        "basemap": {},

        "dataSources": [
            {
                "name": data_source_name,
                "itemType": "KqlDatabase",
                "workspaceId": cfg.workspace_id,
                "itemId": kql_database_item_id
            }
        ],

        "iconSources": [],

        "layerSources": [
            {
                "id": layer_source_id,
                "name": cfg.kql_function_name,
                "type": "kusto",
                "dataSourceName": data_source_name,
                "workspaceId": cfg.workspace_id,
                "itemId": kql_database_item_id,
                "refreshIntervalMs": cfg.refresh_interval_ms,
                "queryType": "function",
                "query": f"{cfg.kql_function_name}()"
            }
        ],
        "layerSettings": [
            {
                "id": layer_setting_id,
                "name": "Live Locations",
                "sourceId": layer_source_id,
                "options": {
                    "type": "vector",
                    "visible": True,
                    "pointLayerType": "bubble",
                    "tooltipKeys": ["VehicleId", "EventTime"],
                    "bubbleOptions": {
                        "color": "#0078D4"
                    }
                },
                "latitudeColumnName": "Latitude",
                "longitudeColumnName": "Longitude"

            }
        ]
    }

    
    print("Map definition (map.json):", json.dumps(map_json, indent=2))
    return map_json

Criar .platform (metadados de plataforma)

build_platform_json cria e retorna uma parte opcional .platform que a chamada Criar Mapa pode incluir ao lado map.json quando você deseja definir metadados de item não padrão no Mapa. A inclusão de uma parte .platform não é necessária – Fabric aplica metadados padrão quando a parte é omitida , mas este tutorial mostra como criar uma para que você possa reutilizar o padrão quando precisar de controle explícito sobre o tipo de item, o nome de exibição, a descrição ou um identificador lógico estável.

O conteúdo segue o esquema de propriedades da plataforma e tem duas seções: metadata (type: "Map", , displayNamedescription) e config (version, logicalId). O logicalId é gerado aqui como um novo UUID, o que é adequado para uma criação única; se você planeja reimplantar o mesmo mapa por meio da integração com o Git ou de execuções repetidas, defina logicalId como um valor estável para que as atualizações sejam aplicadas ao mesmo item.

Para obter mais informações, consulte a definição de item do mapa e a visão geral da definição de item.

Adicione o seguinte após a build_map_json função:

# =========================================================
# Build .platform (platform metadata)
# =========================================================

def build_platform_json(cfg: Config) -> dict:
    """
    Build and return the optional .platform payload for a Fabric Map item.

    The map definition supports an optional .platform part alongside
    map.json that carries non-default item metadata: the item type,
    display name and description, and a `logicalId` used for
    deterministic updates. Fabric applies defaults when the part is
    omitted, so this payload is only needed when you want explicit
    control over those fields. A fresh UUID is used for `logicalId`
    here; pin it to a stable value if repeat runs should target the
    same item.
    """
    return {
        "$schema": "https://developer.microsoft.com/json-schemas/fabric/gitIntegration/platformProperties/2.0.0/schema.json",
        "metadata": {
            "type": "Map",
            "displayName": cfg.map_display_name,
            "description": cfg.map_description
        },
        "config": {
            "version": "2.0",
            # Use a stable logicalId if you want deterministic updates; UUID is fine for create.
            "logicalId": str(uuid.uuid4())
        }
    }

Criar um mapa com definição embutida

create_map cria o mapa enviando por POST a definição embutida que você montou e retorna a ID do item do novo mapa. A solicitação carrega três partes codificadas em base64 em payloadType: "InlineBase64": map.json (a definição de núcleo necessária), os metadados opcionais .platform criados na etapa anterior e um arquivo de consulta Kusto nomeado queries/layerSource-<layerSourceId>.kql que contém a chamada para a função KQL armazenada. Agrupar as três partes em uma única chamada provisiona o mapa e vincula sua camada de dados à função KQL de forma atômica, de modo que nenhuma nova ida e volta de getDefinition / updateDefinition seja necessária.

O nome do arquivo de consulta importa: Fabric resolve a consulta de uma camada combinando queries/layerSource-<layerSourceId>.kql com o id da entrada correspondente em layerSources, de modo que a função extrai a ID de origem da camada de map_json["layerSources"][0]["id"] para construir o caminho. map.json e .platform são codificados em base64 por meio _json_to_b64; o texto da consulta é codificado em base64 diretamente porque é uma cadeia de caracteres em vez de um dict.

A API REST de criação de mapas pode responder com 201 Created (síncrona, ID em linha), 202 Accepted (LRO assíncrona via Location ou x-ms-operation-id) ou 200 OK, com um payload de conclusão contendo apenas o status, em que o mapa ainda não está visível em List Maps devido ao atraso de propagação do backend. _handle_lro abrange todos esses casos , incluindo listagem e correspondência por displayName – de modo que essa função delega o tratamento de resposta completo a ele em uma única chamada.

Para obter mais informações, consulte a definição de item de mapa.

Adicione o seguinte após a build_platform_json função:

# =========================================================
# Create a map with inline definition
# =========================================================


def create_map(client: httpx.Client, fabric: FabricClient, cfg: Config, map_json: dict, platform_json: dict) -> str:
    """
    Create the Fabric Map with its definition inline and return its item ID.

    Sends a single Create Map request whose `parts` array carries three
    base64-encoded payloads: `map.json` (the required core definition),
    the optional `.platform` metadata, and a Kusto query file named
    `queries/layerSource-<layerSourceId>.kql` whose `<layerSourceId>`
    matches `map_json["layerSources"][0]["id"]` so Fabric can bind the
    query to the layer. Delegates response handling to `_handle_lro`,
    which covers synchronous, asynchronous, and status-only completions.
    """

    create_map_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/maps"

    # Extract the layer source id so we can name the query file correctly
    layer_source_id = map_json["layerSources"][0]["id"]

    # Kusto query content (bind to the stored function)
    query_text = f"{cfg.kql_function_name}()"
    query_b64 = base64.b64encode(query_text.encode("utf-8")).decode("utf-8")

    create_map_payload = {
        "displayName": cfg.map_display_name,
        "description": cfg.map_description,
        "definition": {
            "parts": [
                {
                    "path": "map.json",
                    "payload": _json_to_b64(map_json),
                    "payloadType": "InlineBase64"
                },
                {
                    "path": ".platform",
                    "payload": _json_to_b64(platform_json),
                    "payloadType": "InlineBase64"
                },
                {
                    # Kusto layer query file naming convention
                    "path": f"queries/layerSource-{layer_source_id}.kql",
                    "payload": query_b64,
                    "payloadType": "InlineBase64"
                }
            ]
        }
    }

    map_resp = fabric.request("POST", create_map_url, json_body=create_map_payload)

    return _handle_lro(
        client, map_resp,
        list_url=create_map_url,
        match_display_name=cfg.map_display_name,
    )

Orquestrar o fluxo de trabalho

main é o único ponto de entrada que executa o tutorial de ponta a ponta. Ele instancia Config, abre um httpx.Client reutilizado em todos os auxiliares, o encapsula em um FabricClient e, em seguida, chama a função de cada etapa na ordem de dependência: create_eventhouse → create_kql_table_if_missing (deve existir antes que o fluxo de eventos se associe a ele) → create_eventstream_with_definition → seed_eventstream_from_csv → verify_eventhouse_data (detecta uma configuração incorreta de ingestão antes de qualquer trabalho de mapa) → create_kql_function → wait_for_kql_database_ready (barreira best-effort para que Create Map possa resolver o banco de dados KQL) → build_map_json → build_platform_json → create_map.

A ordem importa porque a maioria das etapas consome algo criado por uma etapa anterior — create_eventstream_with_definition precisa do databasesItemIds do eventhouse, e create_map precisa do nome da função KQL e do identificador do item do banco de dados KQL. O bloco print final apresenta as IDs de cada recurso criado para que você possa encontrá-las no portal Fabric.

Adicione o seguinte após a create_map função:

# =========================================================
# main(): orchestrates the full workflow
# =========================================================

def main():
    """
    Orchestrate the tutorial workflow.

    1) Create eventhouse
    2) Create KQL table (required for ingestion)
    3) Create Eventstream (definition-based)
    4) Seed initial data so the map is not empty on first open
    5) Validate ingestion BEFORE moving on
    6) Create KQL function (required for Maps layer)
    7) Ensure KQL database is available to Maps
    8) Build map.json
    9) Build .platform metadata
    10) Create map with inline definition
    """
    cfg = Config()

    print("Initializing clients...")
    with httpx.Client(timeout=60) as client:
        fabric = FabricClient(client)

        # Step 1: Create eventhouse
        eventhouse_id = create_eventhouse(client, fabric, cfg)

        # Step 2: Ensure table exists BEFORE Eventstream binds to it
        create_kql_table_if_missing(client, fabric, cfg, eventhouse_id)

        # Step 3: Create Eventstream (definition-based)
        eventstream_id = create_eventstream_with_definition(client, fabric, cfg, eventhouse_id)

        # Step 4: Seed initial data so the map is not empty on first open
        seed_count = seed_eventstream_from_csv(cfg)

        # Step 5: Validate ingestion BEFORE moving on
        verify_eventhouse_data(client, fabric, cfg, eventhouse_id)

        # Step 6: Create KQL function (required for Maps layer)
        kql_database_item_id = create_kql_function(client, fabric, cfg, eventhouse_id)

        # Step 7: Ensure KQL database is available to Maps
        wait_for_kql_database_ready(client, cfg, kql_database_item_id)

        # Step 8: Build map.json (Kusto function layer)
        map_json = build_map_json(cfg, kql_database_item_id)

        # Step 9: Build .platform metadata
        platform_json = build_platform_json(cfg)

        # Step 10: Create map with inline definition
        map_id = create_map(client, fabric, cfg, map_json, platform_json)

        print("\nDONE")
        print("Eventhouse ID:", eventhouse_id)
        print("Eventstream ID:", eventstream_id)
        print(f"Seed events sent: {seed_count}")
        print("KQL database item ID:", kql_database_item_id)
        print("KQL function:", cfg.kql_function_name)
        print("Map ID:", map_id)

if __name__ == "__main__":
    main()

Executar o aplicativo

Durante a execução do script, você será solicitado a colar a cadeia de conexão do fluxo de eventos.

Para recuperar esse valor:

1. Abra o espaço de trabalho do Fabric
2. Abra o fluxo de eventos criado pelo script
3. Selecione a origem do ponto de extremidade personalizado
4. Abrir autenticação de chave SAS
5. Copiar string de conexão - chave primária

Cole o valor no console quando solicitado.

Executar o script:

python create_realtime_map.py

Verifique se todos os itens foram criados:

Neste ponto, todos os recursos são criados e configurados.

Para simular o streaming contínuo e assistir à atualização do mapa quase em tempo real, continue com o acompanhamento Tutorial: simular a ingestão de dados em tempo real para um mapa usando APIs REST e Python. Ele se baseia diretamente neste tutorial e reutiliza a casa de eventos, o fluxo de eventos, a função KQL e o mapa que você criou.

Resumo

Neste tutorial, você provisionou os recursos necessários para uma solução geoespacial em tempo real em Microsoft Fabric, usando Fabric APIs REST e Python.

Você realizou o seguinte:

• Criou um banco de dados eventhouse e KQL usando a API REST Fabric
• Criou um fluxo de eventos com um ponto de extremidade personalizado para ingerir eventos de streaming
• Definiu uma função KQL para consultar e formatar dados em tempo real para visualização de mapa
• Criou e implantou um mapa Fabric com uma definição embutida referenciando dados do Eventhouse
• Semeou o fluxo de eventos com eventos iniciais para que o mapa exibia dados imediatamente

Essa arquitetura demonstra um padrão de análise em tempo real comum em Fabric:

• Produtores externos enviam eventos para Eventstream
• Eventstream roteia e ingere dados no Eventhouse
• Funções KQL transformam os dados
• Mapas consultam Eventhouse e são atualizados automaticamente para refletir novos eventos

Ao automatizar a criação de recursos usando APIs REST e Python, agora você tem uma abordagem repetível para criar aplicativos espaciais em tempo real sem configuração manual. Para enviar dados continuamente ao mapa, continue com o tutorial complementar do simulador.

Próximas Etapas 

Agora que você entende o fluxo de ponta a ponta, pode estender essa solução para incorporar um simulador em tempo real.

Para obter um tutorial que demonstra a criação de um simulador em tempo real para o mapa que você acabou de criar usando APIs REST, consulte: