Practical examples and integration patterns for the Standard Metrics API
client_id
and client_secret
credentialsrequests
library installedimport requests
import base64
def get_access_token(client_id: str, client_secret: str) -> str:
"""Get access token from Standard Metrics API."""
# Encode credentials
credential = f"{client_id}:{client_secret}"
encoded_credential = base64.b64encode(credential.encode("utf-8")).decode("utf-8")
# Get access token
response = requests.post(
"https://api.standardmetrics.io/o/token/",
headers={
"Authorization": f"Basic {encoded_credential}",
"Content-Type": "application/x-www-form-urlencoded",
},
data={"grant_type": "client_credentials"}
)
response.raise_for_status()
return response.json()["access_token"]
# Usage
client_id = "YOUR_CLIENT_ID"
client_secret = "YOUR_CLIENT_SECRET"
access_token = get_access_token(client_id, client_secret)
headers = {"Authorization": f"Bearer {access_token}"}
def get_all_companies(headers: dict[str, str]) -> list[dict]:
"""Fetch all companies with pagination handling."""
all_companies = []
url = "https://api.standardmetrics.io/v1/companies/"
while url:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
all_companies.extend(data.get("results", []))
url = data.get("next") # Handle pagination
return all_companies
# Usage
companies = get_all_companies(headers)
print(f"Found {len(companies)} companies")
# Create a name-to-ID mapping for easy lookup
company_map = {company["name"]: company["id"] for company in companies}
def create_company(headers: dict[str, str], company_data: dict) -> dict:
"""Create a new company in your portfolio."""
response = requests.post(
"https://api.standardmetrics.io/v1/companies/",
headers=headers,
json=company_data
)
response.raise_for_status()
return response.json()
# Example usage
new_company = {
"name": "Acme Corp",
"website": "https://acme.com",
"description": "Technology company",
"stage": "Series A"
}
company = create_company(headers, new_company)
print(f"Created company '{company['name']}' with ID: {company['id']}")
def get_company_metrics(headers: dict[str, str], company_id: str, **filters) -> list[dict]:
"""Get metrics for a specific company with optional filters."""
params = {"company_id": company_id}
params.update(filters)
response = requests.get(
"https://api.standardmetrics.io/v1/metrics/",
headers=headers,
params=params
)
response.raise_for_status()
return response.json().get("results", [])
# Example: Get revenue and cash metrics for the last year
company_id = "your_company_id_here"
metrics = get_company_metrics(
headers,
company_id,
category=["revenue", "cash_in_bank"],
cadence="month",
from_date="2023-01-01",
to_date="2023-12-31"
)
print(f"Retrieved {len(metrics)} metric records")
# Process the results
for metric in metrics:
print(f"{metric['category']}: ${metric['value']} on {metric['date']}")
from collections import defaultdict
def get_portfolio_metrics(headers: dict[str, str], **filters) -> list[dict]:
"""Get metrics for all companies in portfolio with pagination."""
all_metrics = []
url = "https://api.standardmetrics.io/v1/metrics/"
params = filters
while url:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
all_metrics.extend(data.get("results", []))
url = data.get("next")
params = {} # Clear params for subsequent pages
return all_metrics
# Example: Get latest revenue for all companies
portfolio_metrics = get_portfolio_metrics(
headers,
category="revenue",
cadence="month",
limit=100
)
print(f"Retrieved {len(portfolio_metrics)} metrics across portfolio")
# Group by company
metrics_by_company = defaultdict(list)
for metric in portfolio_metrics:
metrics_by_company[metric["company_id"]].append(metric)
def get_company_documents(headers: dict[str, str], company_id: str = None, **filters) -> list[dict]:
"""Get documents, optionally filtered by company and criteria."""
params = filters.copy()
if company_id:
params["company_id"] = company_id
response = requests.get(
"https://api.standardmetrics.io/v1/documents/",
headers=headers,
params=params
)
response.raise_for_status()
return response.json().get("results", [])
# Get all documents for a specific company
company_id = "your_company_id_here"
documents = get_company_documents(headers, company_id)
print(f"Found {len(documents)} documents for company")
# Get financial statements only
financial_docs = get_company_documents(
headers,
company_id,
document_type="financial_statement"
)
print(f"Found {len(financial_docs)} financial statements")
def upload_document(headers: dict[str, str], document_data: dict) -> dict:
"""Upload a new document to the system."""
response = requests.post(
"https://api.standardmetrics.io/v1/documents/",
headers=headers,
json=document_data
)
response.raise_for_status()
return response.json()
# Example usage
document = {
"company_id": "your_company_id_here",
"title": "Q4 2023 Financial Report",
"document_type": "financial_statement",
"url": "https://example.com/q4-report.pdf",
"date": "2023-12-31"
}
uploaded_doc = upload_document(headers, document)
print(f"Uploaded document '{uploaded_doc['title']}' with ID: {uploaded_doc['id']}")
def push_metrics(headers: dict[str, str], company_id: str, metrics: list[dict]) -> dict:
"""Push new metrics to Standard Metrics."""
payload = {
"company_id": company_id,
"metrics": metrics
}
response = requests.post(
"https://api.standardmetrics.io/v1/metrics/",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
# Example: Push monthly revenue data
metrics_data = [
{
"category": "revenue",
"value": 150000,
"date": "2023-10-31",
"cadence": "month",
"currency": "USD"
},
{
"category": "revenue",
"value": 162000,
"date": "2023-11-30",
"cadence": "month",
"currency": "USD"
}
]
company_id = "your_company_id_here"
result = push_metrics(headers, company_id, metrics_data)
import base64
class StandardMetricsClient:
"""Simple client for Standard Metrics API."""
def __init__(self, client_id: str, client_secret: str):
"""Initialize client with OAuth credentials."""
self.headers = self._get_headers(client_id, client_secret)
def _get_headers(self, client_id: str, client_secret: str) -> dict[str, str]:
"""Get authorization headers."""
access_token = get_access_token(client_id, client_secret)
return {"Authorization": f"Bearer {access_token}"}
def get_companies(self) -> list[dict]:
"""Get all companies in portfolio."""
return get_all_companies(self.headers)
def get_metrics(self, company_id: str, **filters) -> list[dict]:
"""Get metrics for a company with optional filters."""
return get_company_metrics(self.headers, company_id, **filters)
def push_metrics(self, company_id: str, metrics: list[dict]) -> dict:
"""Push new metrics for a company."""
return push_metrics(self.headers, company_id, metrics)
# Usage example
client = StandardMetricsClient("your_client_id", "your_client_secret")
# Get companies
companies = client.get_companies()
print(f"Found {len(companies)} companies")
# Get metrics for first company
if companies:
company = companies[0]
metrics = client.get_metrics(
company["id"],
category="revenue",
cadence="month",
from_date="2023-01-01",
to_date="2023-12-31"
)
print(f"Found {len(metrics)} revenue metrics for {company['name']}")
# Push new metric
new_metrics = [{
"category": "revenue",
"value": 200000,
"date": "2023-12-31",
"cadence": "month",
"currency": "USD"
}]
result = client.push_metrics(company["id"], new_metrics)
import requests
import time
def safe_api_call(func, *args, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, **kwargs) -> requests.Response:
"""
Always raises on failure.
Success: any 2xx -> returns the Response.
Retries: 429, 5xx, Timeout, RequestException (exponential backoff, no jitter).
Fails fast: 3xx (unexpected redirect) and other 4xx except 429.
"""
last_exc = None
for attempt in range(max_retries + 1):
try:
resp = func(*args, **kwargs)
code = resp.status_code
if 200 <= code < 300:
return resp
# Retry on 429 and 5xx
if code == 429 or 500 <= code < 600:
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
time.sleep(delay)
continue
resp.raise_for_status()
resp.raise_for_status()
except (requests.exceptions.Timeout, requests.exceptions.RequestException) as e:
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
time.sleep(delay)
continue
raise
raise RuntimeError("safe_api_call: exhausted attempts without a response")
# Usage example
resp = safe_api_call(
requests.get,
"https://api.standardmetrics.io/v1/companies/",
headers=headers,
timeout=30
)
result = resp.json()
companies = result.get("results", [])
print(f"Successfully fetched {len(companies)} companies")
import time
from functools import wraps
def rate_limit(calls_per_minute: int = 450):
"""Decorator to add rate limiting to functions."""
min_interval = 60.0 / calls_per_minute
last_called = [0.0] # Use list to allow modification in nested function
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
if elapsed < min_interval:
sleep_time = min_interval - elapsed
time.sleep(sleep_time)
last_called[0] = time.time()
return func(*args, **kwargs)
return wrapper
return decorator
# Usage with decorator
@rate_limit(calls_per_minute=400) # Stay under the 500/min limit
def fetch_company_data(company_id: str, headers: dict[str, str]) -> dict:
"""Fetch company data with automatic rate limiting."""
response = requests.get(
f"https://api.standardmetrics.io/v1/companies/{company_id}/",
headers=headers
)
response.raise_for_status()
return response.json()
# Usage example
for company_id in company_ids:
company_data = fetch_company_data(company_id, headers)
print(f"Fetched data for {company_data['name']}")