Este producto no es compatible con el sitio Datadog seleccionado. ().

La acción de función de Python le permite escribir scripts de Python personalizados para transformaciones de datos, parseo y enriquecimiento de carga útil dentro de sus flujos de trabajo.

Entorno de Python

La acción de función de Python se ejecuta en un entorno de ejecución restringido con las siguientes características:

  • Network access: Restricted
  • Python version: 3.12.8
  • Available libraries: In addition to the Python standard library, the following packages are available:
    PackageVersionDescription
    python-dateutil==2.9.0Extensions to the standard Python datetime module
    python-hcl2==7.3.1A parser for HCL2
    PyYAML==6.0.3YAML parser and emitter for Python
    rsa==4.9.1Pure-Python RSA implementation

Estructura del script

Todos los scripts de Python deben definir una función main que acepte un parámetro ctx de tipo Context. Por ejemplo:

from execution_context import Context

def main(*, ctx: Context):
  # Use ctx to access Trigger or Steps data
  workflow_name = ctx["WorkflowName"]
  return f"Running workflow {workflow_name!r}"

El objeto ctx proporciona acceso a todas las variables de contexto del flujo de trabajo, similar a la variable $ en las expresiones de JavaScript. Utilice el acceso de estilo diccionario (por ejemplo, ctx["Steps"]["Step_name"]["variable"]) para hacer referencia a valores de pasos anteriores.

Agregar una acción de función de Python

En el lienzo del flujo de trabajo:

  1. Haga clic en + para agregar un paso al flujo de trabajo.
  2. Busque Python.
  3. Seleccione la acción de Python para agregarla a su flujo de trabajo.

Escribir scripts de Python con IA

Puede usar Bits AI para ayudar a escribir scripts de Python dentro de un paso del flujo de trabajo.

Para escribir un script con Bits AI:

  1. Agregue un paso a su flujo de trabajo.
  2. En la sección Inputs, haga clic en Write Code with AI.
  3. Ingrese un prompt personalizado o seleccione uno de los prompts de ejemplo.
  4. Opcionalmente, haga clic en Test script para generar una vista previa del paso del flujo de trabajo.
  5. Para guardar el script, haga clic en Accept changes. Para continuar editando el script, haga clic en Reject changes.
  6. Haga clic en X para cerrar el cuadro de diálogo de IA.
  7. Ingrese un Description.
  8. Haga clic en Save.

Ejemplos de scripts

Analizar y transformar datos JSON

Este ejemplo analiza una cadena JSON de un paso anterior y extrae campos específicos.

from execution_context import Context
import json

def main(*, ctx: Context):
    # Get JSON string from previous step
    json_string = ctx["Steps"]["Get_data"]["output"]

    # Parse and transform
    data = json.loads(json_string)
    return {
        "user_ids": [user["id"] for user in data["users"]],
        "total_count": len(data["users"])
    }

Trabajar con fechas y marcas de tiempo

Este ejemplo utiliza la biblioteca python-dateutil para realizar cálculos de fechas.

from execution_context import Context
from dateutil import parser, relativedelta
from datetime import datetime

def main(*, ctx: Context):
    # Parse a date string
    start_date = parser.parse(ctx["Trigger"]["date_string"])

    # Calculate date 30 days in the future
    future_date = start_date + relativedelta.relativedelta(days=30)

    return {
        "start": start_date.isoformat(),
        "end": future_date.isoformat(),
        "days_difference": 30
    }

Operaciones criptográficas

Este ejemplo utiliza la biblioteca rsa para cifrar un mensaje.

from execution_context import Context
import rsa
import base64

def main(*, ctx: Context):
    # Get message from workflow context
    message = ctx["Steps"]["Compose_message"]["text"]

    # Generate RSA key pair
    (public_key, private_key) = rsa.newkeys(512)

    # Encrypt message
    encrypted = rsa.encrypt(message.encode(), public_key)

    return {
        "encrypted_message": base64.b64encode(encrypted).decode(),
        "public_key": public_key.save_pkcs1().decode()
    }