Python Scripts¶
Call the LightPane API from Python to pull cloud resource data into scripts, reports, and automation workflows.
Requirements:
- Python 3.7+
requestslibrary (pip install requests)- A LightPane API key (
csl_ak_prefix) — create one here
Discover a single service¶
discover_ec2.py
import requests
API_URL = "https://api.lightpane.cloud/services/discover"
TOKEN = "csl_ak_YOUR_API_KEY_HERE"
response = requests.post(API_URL, json={
"services": [{
"service": "ec2",
"provider": "aws",
"region": "eu-west-2",
"attributes": ["name", "instance_id", "instance_type", "state", "private_ip"]
}]
}, headers={
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
})
response.raise_for_status()
data = response.json()
for result in data.get("results", []):
print(f"\n{result['metadata']['service_label']}")
print(f"{'—' * 60}")
for row in result.get("rows", []):
print(row)
Run it:
Discover multiple services in one call¶
discover_batch.py
import requests
API_URL = "https://api.lightpane.cloud/services/discover"
TOKEN = "csl_ak_YOUR_API_KEY_HERE"
response = requests.post(API_URL, json={
"services": [
{
"service": "ec2",
"provider": "aws",
"region": "eu-west-2",
"attributes": ["name", "instance_id", "state"]
},
{
"service": "s3",
"provider": "aws",
"region": "eu-west-2",
"attributes": ["bucket_name", "creation_date", "region"]
},
{
"service": "lambda_functions",
"provider": "aws",
"region": "eu-west-2",
"attributes": ["name", "runtime", "memory", "last_modified"]
}
]
}, headers={
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
})
response.raise_for_status()
data = response.json()
for result in data.get("results", []):
service = result["metadata"]["service_label"]
rows = result.get("rows", [])
print(f"\n{service} ({len(rows)} resources)")
for row in rows:
print(f" {row}")
Parse the response¶
The API returns a JSON object with this structure:
{
"success": true,
"results": [
{
"metadata": {
"service": "ec2",
"service_label": "EC2 Instances",
"provider": "aws",
"region": "eu-west-2",
"resource_count": 3
},
"columns": [
{"id": "name", "label": "Name"},
{"id": "instance_id", "label": "Instance ID"},
{"id": "state", "label": "State"}
],
"rows": [
{"name": "web-server-1", "instance_id": "i-0abc123", "state": "running"},
{"name": "api-server", "instance_id": "i-0def456", "state": "running"},
{"name": "worker", "instance_id": "i-0ghi789", "state": "stopped"}
]
}
]
}
Each entry in results corresponds to one service request. columns describes the
available fields. rows contains the resource data.
Export to CSV¶
export_csv.py
import csv
import requests
API_URL = "https://api.lightpane.cloud/services/discover"
TOKEN = "csl_ak_YOUR_API_KEY_HERE"
response = requests.post(API_URL, json={
"services": [{
"service": "ec2",
"provider": "aws",
"region": "eu-west-2",
"attributes": ["name", "instance_id", "instance_type", "state", "private_ip"]
}]
}, headers={
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
})
response.raise_for_status()
data = response.json()
for result in data.get("results", []):
columns = [col["id"] for col in result["columns"]]
rows = result.get("rows", [])
with open("ec2_instances.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=columns)
writer.writeheader()
writer.writerows(rows)
print(f"Wrote {len(rows)} rows to ec2_instances.csv")
Use an environment variable for the token¶
Do not hard-code the token in scripts that are committed to version control.
discover_env.py
import os
import requests
API_URL = "https://api.lightpane.cloud/services/discover"
TOKEN = os.environ["LIGHTPANE_API_KEY"]
response = requests.post(API_URL, json={
"services": [{
"service": "ec2",
"provider": "aws",
"region": "eu-west-2"
}]
}, headers={
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json"
})
response.raise_for_status()
print(response.json())
Handle errors¶
response = requests.post(API_URL, json={...}, headers={...})
if response.status_code == 401:
print("Token is invalid or expired. Create a new access key.")
elif response.status_code == 403:
error = response.json().get("error", "Forbidden")
print(f"Access denied: {error}")
elif response.status_code == 429:
print("Rate limit exceeded. Wait and retry.")
elif response.status_code >= 400:
print(f"Error {response.status_code}: {response.text}")
else:
data = response.json()
# process data
| Status | Meaning |
|---|---|
| 200 | Discovery completed |
| 401 | Invalid or expired token |
| 403 | Constraint violation (origin, IP, service scope) |
| 429 | Rate limit exceeded |
| 500 | Server error |