Pular para conteúdo

API — BigQuery e Dataform

📋 Visão Geral

A camada de API do Motor de Pricing é responsável por toda comunicação com o Google Cloud Platform. Ela compreende:

  • Leitura de dados via BigQuery (queries parametrizadas)
  • Escrita de dados processados de volta ao BigQuery
  • Orquestração de pipelines via Dataform

🔄 Fluxo de Dados

graph LR
    subgraph Leitura
        A[fetch_data.py] -->|SELECT| B[BigQuery - gold_pricing]
    end

    subgraph Processamento
        C[Usuário configura na UI] --> D[send_data.py]
        D -->|INSERT| B
        D --> E[dataform_trigger.py]
        E -->|Trigger| F[Pipeline Dataform]
        F -->|Processa| B
    end

    subgraph Resultado
        B --> G[fetch_data.py]
        G --> H[Tabela de Preços na UI]
    end

    style B fill:#4285F4,color:#fff
    style F fill:#34A853,color:#fff

📦 Módulos da API

bigquery_client.py — Cliente Centralizado

Provê a instância do cliente BigQuery utilizada por todos os outros módulos. Centraliza configuração de projeto e credenciais.

fetch_data.py — Leitura de Dados

838 linhas — módulo mais extenso da API.

Aspecto Detalhe
Função principal fetch_data(table_name, filter, user_id)
Projeto GCP GCP_PROJECT_ID (dev) ou driven-strength-374020 (prod)
Dataset GCP_DATASET_ID (dev) ou gold_pricing (prod)
Performance Suporte a ThreadPoolExecutor para fetches paralelos

Tabelas consultadas incluem:

Tabela Uso
parametrization Parametrizações de data
cost Custos dos produtos
gross_margin Margem bruta
products Cadastro de produtos
packaging Perfis de embalagem
regionalization Perfis regionais
current_price_list Lista de preços vigente
product_strategy Estratégia de produtos
product_strategy_architecture Arquitetura de produtos
regionalization_strategy Estratégia de regionalização
rotated_market Mercado rotacionado
vs_sales_suggestion VS Sugestão de vendas
caps_vs_last_price_list CAPS vs última lista
min_margin Margem mínima
base_index Índice base
painel_qualidade_alertas Alertas de qualidade
cc_posicionamento_produtos Posicionamento de produtos
price_decomposition Decomposição do preço

send_data.py — Envio de Parametrização

Envia os dados configurados pelo usuário na Arquitetura de Preços para o BigQuery.

Aspecto Detalhe
Função principal send_data(parametrization_data, variables_data, user_id)
Retorno {"status": True/False, "message": "...", "request_id": "UUID"}
Processo 1. Deserializa JSON → 2. Aplica schema → 3. Limpa DataFrame → 4. Filtra colunas → 5. Insere no BigQuery → 6. Trigger Dataform
Auxiliar filter_dataframe_columns() — remove colunas que não existem no schema da tabela destino

save_price_list.py — Salvamento de Lista de Preços

Salva a lista de preços final no BigQuery, filtrando apenas os itens marcados para salvar.

Aspecto Detalhe
Função principal save_price_list(data, price_list_id, user_id)
Filtro Apenas linhas com salvar_preco == "Sim"
Auxiliar filter_and_prepare_data() — adiciona price_list_id e user_id
Auxiliar clean_dataframe_for_parquet() — corrige tipos mistos, objetos aninhados e NaN

send_preco_ponta_data.py — Preço Ponta

Envia dados de parametrização para a tabela de preço ponta, dispara o pipeline Dataform e retorna os dados processados.

Aspecto Detalhe
Função principal send_preco_ponta_data(parametrization_data, user_id)
Retorno {"status": True/False, "message": "...", "request_id": "UUID", "data": [...]}
Diferença Diferente do send_data, este módulo retorna os dados gerados pela pipeline

dataform_trigger.py — Orquestração de Pipelines

601 linhas — Classe DataformTrigger que orquestra execuções do Dataform.

sequenceDiagram
    participant App as Motor Pricing
    participant DF as Dataform API
    participant BQ as BigQuery

    App->>DF: trigger_pipeline(pipeline_name)
    DF->>DF: Criar CompilationResult
    DF->>DF: Criar WorkflowInvocation
    App->>DF: wait_for_workflow_completion(timeout=300s)
    loop A cada 10s
        App->>DF: get_workflow_status()
        DF-->>App: Status (RUNNING/SUCCEEDED/FAILED)
    end
    DF->>BQ: Executa transformações
    BQ-->>App: Dados processados
Método Finalidade
__init__(project_id, location, repository_id) Inicializa cliente Dataform
trigger_pipeline(pipeline_name) Compila e executa pipeline
get_workflow_status(workflow_name) Consulta status de execução
wait_for_workflow_completion(timeout=300) Monitora com polling a cada 10s
list_compilation_actions(compilation_result) Lista ações disponíveis
list_workspaces() Lista workspaces do repositório
get_repository_info() Informações do repositório
trigger_pipeline_with_workspace(workspace) Executa via workspace específico

schema_utils.py — Validação de Schema

Utilitários para garantir que os DataFrames estejam compatíveis com os schemas das tabelas BigQuery antes do envio.

dataframe_utils.py — Limpeza de DataFrames

Funções para limpeza e preparação de DataFrames para formato Parquet, corrigindo problemas comuns como tipos mistos, objetos aninhados e NaN em colunas de texto.

Última atualização: Março 2026 | Versão 1.0