Create and manage reports

1
import requests
2
3
BASE_URL = "https://alfa.boosted.ai/client"
4
API_KEY = "YOUR_API_KEY_HERE"
5
6
headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
7
8
def run_workplan(workplan_id):
9
    """Execute a workplan to generate a new report."""
10
    url = f"{BASE_URL}/v2/workplans/{workplan_id}/run-workplan"
11
    
12
    response = requests.post(url, headers=headers)
13
    
14
    if response.status_code == 200:
15
        data = response.json()
16
        report_id = data.get("report_id")
17
        status = data.get("status")
18
        print(f"Workplan execution started! Report ID: {report_id}")
19
        print(f"Current status: {status}")
20
        return report_id
21
    else:
22
        print(f"Error: {response.status_code}, {response.text}")
23
        return None
24
25
# Run a workplan to generate a new report
26
workplan_id = "your-workplan-id"
27
report_id = run_workplan(workplan_id)

1
import time
2
3
def get_report_status(report_id):
4
    """Check the status of a report generation."""
5
    url = f"{BASE_URL}/v2/reports/{report_id}/results"
6
    
7
    response = requests.get(url, headers=headers)
8
    
9
    if response.status_code == 200:
10
        data = response.json()
11
        status = data.get("status")
12
        created_at = data.get("created_at")
13
        print(f"Report {report_id}: {status} (created: {created_at})")
14
        return status
15
    else:
16
        print(f"Error: {response.status_code}, {response.text}")
17
        return None
18
19
# Poll until the report is complete
20
report_status = ""
21
while report_status not in ["COMPLETE", "ERROR", "CANCELLED"]:
22
    report_status = get_report_status(report_id)
23
    
24
    if report_status in ["ERROR", "CANCELLED", "NO_RESULTS_FOUND"]:
25
        print(f"Report execution encountered an issue: {status}")
26
        break
27
    if report_status != "COMPLETE":
28
        print("Report still processing... status: {report_status}")
29
        time.sleep(10)

1
def get_report_results(report_id):
2
    """Get the complete results of a generated report."""
3
    url = f"{BASE_URL}/v2/reports/{report_id}/results"
4
    
5
    response = requests.get(url, headers=headers)
6
    
7
    if response.status_code == 200:
8
        return response.json()
9
    else:
10
        print(f"Error: {response.status_code}, {response.text}")
11
        return None
12
13
# Get the complete report results
14
report_results = get_report_results(report_id)

1
# Get the complete report results
2
report_results = get_report_results(report_id)
3
4
if report_results and "outputs" in report_results:
5
    for i, item in enumerate(report_results["outputs"]):
6
        output_data = item["output"]
7
        output_type = output_data["output_type"]
8
        
9
        print(f"\n--- Output {i+1} ({output_type}) ---")
10
        
11
        if output_type == "text":
12
            # Save the text analysis to a file
13
            with open(f"analysis_{i+1}.txt", "w") as f:
14
                f.write(output_data["val"])
15
            print(f"Text analysis saved to analysis_{i+1}.txt")
16
            
17
        elif output_type == "table":
18
            # Export the table to CSV
19
            import csv
20
            
21
            columns = [col["name"] for col in output_data["columns"]]
22
            with open(f"table_{i+1}.csv", "w", newline="") as f:
23
                writer = csv.writer(f)
24
                writer.writerow(columns)
25
                writer.writerows(output_data["rows"])
26
            print(f"Table data exported to table_{i+1}.csv")
27
            
28
        elif output_type == "graph":
29
            # Just print info about the graph (in a real app, you might render it)
30
            graph_type = output_data["graph"]["graph_type"]
31
            title = output_data["title"]
32
            print(f"Graph: {title} ({graph_type})")

1
def cancel_report(report_id):
2
    """Cancel a report that is currently being generated."""
3
    url = f"{BASE_URL}/v2/reports/{report_id}/cancel"
4
    
5
    response = requests.post(url, headers=headers)
6
    
7
    if response.status_code == 200:
8
        data = response.json()
9
        success = data.get("success", False)
10
        if success:
11
            print(f"Report {report_id} cancelled successfully")
12
        return success
13
    else:
14
        print(f"Error: {response.status_code}, {response.text}")
15
        return False
16
17
# Cancel a report if needed
18
cancel_report(report_id)

1
def list_all_reports():
2
    """List all reports across all workplans."""
3
    url = f"{BASE_URL}/v2/reports"
4
    
5
    response = requests.get(url, headers=headers)
6
    
7
    if response.status_code == 200:
8
        data = response.json()
9
        reports = data.get("reports", [])
10
        
11
        print(f"Total reports: {len(reports)}")
12
        
13
        for report in reports:
14
            print(f"\nReport {report["report_id"]}:")
15
            print(f"  Workplan: {report["workplan_id"]}")
16
            print(f"  Status: {report["status"]}")
17
            print(f"  Created: {report["created_at"]}")
18
        
19
        return reports
20
    else:
21
        print(f"Error: {response.status_code}, {response.text}")
22
        return None
23
24
# List all reports
25
all_reports = list_all_reports()