Voltar para lista de artigos

Pipeline de deploy com Python, Cloud Build, Cloud Run Functions e Google Chat

Deploy de função Python no Cloud Run Functions com Cloud Build

Como engenheiro de dados, lido constantemente com a criação de funções Python para gestão dos processos da área.

Recentemente precisei padronizar o envio de alertas para Google Chat, pois já tenho várias rotinas Python publicadas e rodando diariamente. Cada uma das Cloud Functions, tinha sua própria função interna que executava o envio das mensagens para o Space da equipe.

Com o crescimento da área, volume de dados e número de processos a monitorar, a ideia foi centralizar isso em uma funcao HTTP, que seria chamada por qualquer uma das funções internas, com os parâmetros predefinidos. Para ganhar mais produtividade, o deploy no GCP foi automatizado com Cloud Build.

Neste post vou mostrar, de forma direta, como fazer isso com Cloud Run Functions (2a geracao). ## Estrutura mínima do projeto

send-message/
  main.py
  requirements.txt
  cloudbuild.yaml

Código da função (Cloud Run Functions)

Crie o arquivo main.py:

import os
import requests
from flask import Request, jsonify
def send_message(request: Request):
    payload = request.get_json(silent=True) or {}
    message_data = payload.get("message_data", payload)
    required_fields = ["project", "process", "function", "message_type", "message"]
    missing = [field for field in required_fields if not message_data.get(field)]
    if missing:
        return (
            jsonify(
                {
                    "ok": False,
                    "error": f"Campos obrigatórios ausentes: {', '.join(missing)}",
                }
            ),
            400,
        )
    webhook_url = os.getenv("WEBHOOK_URL")
    if not webhook_url:
        return jsonify({"ok": False, "error": "WEBHOOK_URL não configurada"}), 500
    text = (
        f"[{message_data['message_type']}] "
        f"Projeto: {message_data['project']} | "
        f"Processo: {message_data['process']} | "
        f"Função: {message_data['function']} | "
        f"Mensagem: {message_data['message']}"
    )
    response = requests.post(
        webhook_url,
        json={"text": text},
        headers={"Content-Type": "application/json"},
        timeout=15,
    )
    if response.ok:
        return jsonify({"ok": True, "message": "Notificação enviada"}), 200
    return (
        jsonify(
            {
                "ok": False,
                "error": f"Falha no webhook: {response.status_code} {response.text}",
            }
        ),
        502,
    )

Crie o requirements.txt:

requests==2.32.3

Deploy via Cloud Build

Crie o cloudbuild.yaml:

steps:
  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
    entrypoint: "gcloud"
    args:
      - "functions"
      - "deploy"
      - "send-message"
      - "--gen2"
      - "--region=us-east1"
      - "--runtime=python312"
      - "--source=."
      - "--entry-point=send_message"
      - "--trigger-http"
      - "--allow-unauthenticated"
      - "--set-env-vars=WEBHOOK_URL=${_WEBHOOK_URL}"
timeout: "900s"
options:
  logging: CLOUD_LOGGING_ONLY

Agora rode o build manual para validar:

gcloud builds submit \
  --config cloudbuild.yaml \
  --substitutions _WEBHOOK_URL="https://chat.googleapis.com/v1/spaces/SEU_WEBHOOK"

Na empresa não fazemos os deploys manuais, mas sim via integração direta com o Bitbucket. Em um próimo post, vou detalhar a integração entre o o Bitbucket e o Cloud Build, para que você tenha uma 'esteira' de deploys bem organizada.

Teste rápido via curl

Com a URL da funcao deployada:

curl -X POST "URL_DA_FUNCAO" \
  -H "Content-Type: application/json" \
  -d '{
    "message_data": {
      "project": "NOME_DO_PROJETO",
      "process": "NOME DO PROCESSO",
      "function": "python_function",
      "message_type": "ALERTA",
      "message": "Teste via curl"
    }
  }'

Fechando

Com esse fluxo, você ganha:

  • deploy versionado no repositório;
  • padronizacao de alerta em uma unica funcao;
  • chamada simples via HTTP em qualquer rotina Python.

Se for producao, recomendo trocar --allow-unauthenticated por controle de IAM e mover o webhook para Secret Manager.