sync_job_workflow.py

#!/usr/bin/env python3
"""
JupiterOne Sync Job Workflow Example

This example demonstrates the core synchronization job workflow:
1. Start a sync job
2. Upload entities batch
3. Finalize the sync job

This is a standalone script that can be run independently to test
the sync job functionality.
"""

import os
import sys
from jupiterone.client import JupiterOneClient

def main():
    """Main function to demonstrate sync job workflow."""
    
    # Initialize JupiterOne client
    # You can set these as environment variables or replace with your values
    api_token = os.getenv('JUPITERONE_API_TOKEN')
    account_id = os.getenv('JUPITERONE_ACCOUNT_ID')
    
    if not api_token or not account_id:
        print("Error: Please set JUPITERONE_API_TOKEN and JUPITERONE_ACCOUNT_ID environment variables")
        print("Example:")
        print("  export JUPITERONE_API_TOKEN='your-api-token'")
        print("  export JUPITERONE_ACCOUNT_ID='your-account-id'")
        sys.exit(1)
    
    # Create JupiterOne client
    j1 = JupiterOneClient(api_token=api_token, account_id=account_id)
    
    print("=== JupiterOne Sync Job Workflow Example ===\n")
    
    # You'll need to replace this with an actual integration instance ID
    instance_id = os.getenv('JUPITERONE_INSTANCE_ID')
    if not instance_id:
        print("Error: Please set JUPITERONE_INSTANCE_ID environment variable")
        print("Example:")
        print("  export JUPITERONE_INSTANCE_ID='your-integration-instance-id'")
        sys.exit(1)
    
    try:
        # Step 1: Start sync job
        print("1. Starting synchronization job...")
        sync_job = j1.start_sync_job(
            instance_id=instance_id,
            sync_mode="PATCH",
            source="integration-external"
        )
        
        sync_job_id = sync_job['job'].get('id')
        print(f"✓ Started sync job: {sync_job_id}")
        print(f"  Status: {sync_job['job']['status']}")
        print()
        
        # Step 2: Upload entities batch
        print("2. Uploading entities batch...")
        
        # Sample entities payload
        entities_payload = [
            {
                "_key": "example-server-001",
                "_type": "example_server",
                "_class": "Host",
                "displayName": "Example Server 001",
                "hostname": "server-001.example.com",
                "ipAddress": "192.168.1.100",
                "operatingSystem": "Linux",
                "tag.Environment": "development",
                "tag.Team": "engineering",
                "tag.Purpose": "web_server"
            },
            {
                "_key": "example-server-002",
                "_type": "example_server",
                "_class": "Host",
                "displayName": "Example Server 002",
                "hostname": "server-002.example.com",
                "ipAddress": "192.168.1.101",
                "operatingSystem": "Linux",
                "tag.Environment": "staging",
                "tag.Team": "engineering",
                "tag.Purpose": "database_server"
            },
            {
                "_key": "example-database-001",
                "_type": "example_database",
                "_class": "Database",
                "displayName": "Example Database 001",
                "databaseName": "app_db",
                "engine": "postgresql",
                "version": "13.4",
                "tag.Environment": "development",
                "tag.Team": "data"
            }
        ]
        
        # Upload entities
        upload_result = j1.upload_entities_batch_json(
            instance_job_id=sync_job_id,
            entities_list=entities_payload
        )
        print(f"✓ Uploaded {len(entities_payload)} entities successfully")
        print(f"  Upload result: {upload_result}")
        print()
        
        # Step 3: Finalize sync job
        print("3. Finalizing synchronization job...")
        finalize_result = j1.finalize_sync_job(instance_job_id=sync_job_id)
        
        finalize_job_id = finalize_result['job'].get('id')
        print(f"✓ Finalized sync job: {finalize_job_id}")
        print(f"  Status: {finalize_result['job']['status']}")
        
        # Check final status
        if finalize_result['job']['status'] == 'COMPLETED':
            print("✓ Sync job completed successfully!")
        elif finalize_result['job']['status'] == 'FAILED':
            error_msg = finalize_result['job'].get('error', 'Unknown error')
            print(f"✗ Sync job failed: {error_msg}")
        else:
            print(f"ℹ Sync job status: {finalize_result['job']['status']}")
        
        print("\n=== Sync Job Workflow Complete ===")
        
    except Exception as e:
        print(f"✗ Error during sync job workflow: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()