ログ¶

import json
import datarobot as dr
from datarobot.models.otel import OtelLogEntry

# Initialize DataRobot client for authentication
dr_client = dr.Client()

# Provide DataRobot entity information
entity_type: str = "deployment"
entity_id: str = "<DEPLOYMENT_ID>" # From the deployment Overview tab or URL

# Set output filename
output_filename: str = "datarobot_otel_logs.json"

def export_logs_to_json(logs: list[OtelLogEntry], filename: str) -> None:
    """Export logs to a JSON file"""
    log_entries = []
    for log in logs:
        log_payload = dict(log.__dict__)
        log_payload.update(
            {
                "entity_type": entity_type,
                "entity_id": entity_id,
                "timestamp": log.timestamp.isoformat(),
            }
        )
        log_entries.append(log_payload)

    with open(filename, 'w') as f:
        json.dump(log_entries, f, indent=2)

# Get and export OTel logs
try:
    print(f"Getting logs for {entity_type} {entity_id}...")
    logs: list[OtelLogEntry] = OtelLogEntry.list(entity_type, entity_id)
    print(f"Found {len(logs)} logs")

    # Preview first few logs
    for log in logs[:5]:
        print(f"{log}")

    # Example: Filter for ERROR logs only
    error_logs = [log for log in logs if log.level.upper() == 'ERROR']
    print(f"Found {len(error_logs)} ERROR logs")

    # Export to JSON file
    print(f"Writing logs to {output_filename}...")
    export_logs_to_json(logs, output_filename)
    print(f"Exported {len(logs)} logs to {output_filename}")

except dr.errors.ClientError as e:
    print(f"API Error: {e}")
except FileNotFoundError as e:
    print(f"File Error: {e}")
except Exception as e:
    print(f"Unexpected Error: {e}")