Python 표현식

이 제품은 선택한 Datadog 사이트에서 지원되지 않습니다. ().

Python 함수 작업을 통해 워크플로 내에서 데이터 변환, 구문 분석, 페이로드 보강을 위한 사용자 지정 Python 스크립트를 작성할 수 있습니다.

Python 환경

Python 함수 작업은 다음 특성을 지닌 제한된 실행 환경에서 실행됩니다.

  • 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

스크립트 구조

모든 Python 스크립트는 main 유형의 ctx 파라미터를 허용하는 Context 함수를 정의해야 합니다. 예를 들면 다음과 같습니다.

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

ctx 객체는 JavaScript 표현식의 $ 변수와 유사하게 모든 워크플로 컨텍스트 변수에 대한 액세스 권한을 부여합니다. 사전 스타일 액세스(예: ctx["Steps"]["Step_name"]["variable"])를 사용하여 이전 단계의 값을 참조하세요.

Python 함수 작업 추가

워크플로 캔버스에서 다음을 실행합니다.

  1. +을 클릭하여 워크플로 단계를 추가합니다.
  2. Python을 검색합니다.
  3. Python 작업을 선택하여 워크플로에 추가합니다.

AI로 Python 스크립트 작성

워크플로 단계 내에서 Python 스크립트 작성 시 Bits AI의 도움을 받을 수 있습니다.

Bits AI로 스크립트 작성하기:

  1. 워크플로에 Python 단계를 추가합니다.
  2. Inputs 섹션에서 Write Code with AI를 클릭합니다.
  3. 사용자 지정 프롬프트를 입력하거나 샘플 프롬프트 중 하나를 선택합니다.
  4. (선택 사항) Test script를 클릭하여 워크플로 단계의 미리 보기를 생성합니다.
  5. 스크립트를 저장하려면 Accept changes를 클릭합니다. 스크립트 편집을 계속하려면 Reject changes를 클릭합니다.
  6. X를 클릭하여 AI 대화 상자를 닫습니다.
  7. Description를 입력합니다.
  8. Save를 클릭합니다.

스크립트 예시

JSON 데이터 파싱 및 변환

이 예시에서는 이전 단계의 JSON 문자열을 구문 분석하고 특정 필드를 추출합니다.

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

날짜 및 타임스탬프 작업

이 예시에서는 python-dateutil 라이브러리를 사용하여 날짜 계산을 수행합니다.

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
    }

암호화 작업

이 예시에서는 rsa 라이브러리를 사용하여 메시지를 암호화합니다.

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