5. How to interface with the NRAO ODS API
5.1. Authentication
These are the instructions to authenticate and consume mitigation requests (MR’s) from the NRAO ODS API. All NRAO ODS API endpoints except for health checks require an authentication token.
Contact an NRAO ODS Admin (if you haven’t already) to generate a client for you with appropriate scopes. Coordinate the cadence at which your token will expire with NRAO.
Post your client id and client secret as form data to the NRAO ODS API to get your authentication token. You’ll request this token once and reuse it for all API calls until it expires.
curl -X 'POST' \
'https://ods.nrao.edu/token/' \
-H 'accept: application/json' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'grant_type=client_credentials&client_id=<client_id>&client_secret=<client_secret>'
Response 200 OK
{
"token": "EXAMPLE_TOKEN"
}
3. Pass in this bearer token to authenticate subsequent NRAO ODS API requests. If your token has expired, you will receive a 403, and can repeat step 2 to generate a new token.
curl -X 'GET' \
'https://ods.nrao.edu/ods_data/' \
-H 'accept: application/json' \
-H 'Authorization: Bearer <EXAMPLE_TOKEN>'
5.2. Uploading TLE Data
These are the instructions to authenticate and upload TLE files to the NRAO ODS API.
Contact an NRAO ODS Admin (if you haven’t already) to generate a client for you with appropriate scopes. Coordinate the cadence at which your token will expire with NRAO.
Post your client_id and client_secret as form data to the NRAO ODS API to get your authentication token. You’ll request this token once and reuse it for all API calls until it expires.
curl -X 'POST' \
'https://ods.nrao.edu/token/' \
-H 'accept: application/json' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'grant_type=client_credentials&client_id=<client_id>&client_secret=<client_secret>'
Response 200 OK
{
"token": "<EXAMPLE_TOKEN>"
}
Send your TLE .txt file to the NRAO ODS API using the multipart/form-data content type. You’ll need to include:
Your authentication token in the Authorization header
Your TLE file (a .txt file with TLE records in 3LE format) attached under the field name tle_file
TLE records must be formatted appropriately and a single TLE file must be less than 50 kB.
curl -X 'POST' \
'https://ods.nrao.edu/tle_data' \
-H 'accept: application/json' \
-H 'Authorization: Bearer <EXAMPLE_TOKEN>' \
-H 'Content-Type: multipart/form-data' \
-F 'tle_file=@<EXAMPLE_TEXT_FILE_NAME>.txt;type=text/plain'
Response 201 CREATED
{
"ingested_records": "<number of TLE records sent>"
}
4. When your token expires, you will need to repeat Step 2 using the same client_id and client_secret to get a new token. Your client_id and client_secret do not change; you’ll only need to refresh the token itself.
Option 1: Proactive refresh. Keep track of when you first requested your token and refresh it before it expires.
Option 2: Reactive refresh. If you get a 403 (Forbidden) upon trying to upload TLE data, request a new token from /token and retry the TLE upload with the new token.
5.3. Example Python Script
"""
Example ODS API Client
Usage, to get ODS Data:
python example_ods_api_client.py
To additionally upload a TLE file:
python example_ods_api_client.py --tle_file_path <path to TLE file>
"""
# --------------------------------------------------------------------------------------------------
# NOTE: For demo purposes, loads client credentials from an .env file and persists an
# access token back to the .env file to be used for subsequent requests.
#
# For production, use a secure credential store like your OS keychain or secrets manager instead.
#
# Use the ODS API responsibly. Avoid unnecessary or repeated requests, rotate tokens when needed
# and respect rate limits.
# --------------------------------------------------------------------------------------------------
import argparse
import os
import time
from io import BufferedReader
from typing import Callable, Dict, Optional
import requests
from dotenv import load_dotenv, set_key
from loguru import logger
API_BASE = "https://ods.nrao.edu"
TOKEN_URL = f"{API_BASE}/token"
ODS_DATA_URL = f"{API_BASE}/ods_data"
TLE_DATA_URL = f"{API_BASE}/tle_data"
MAX_REQUEST_RETRIES = 3
RETRY_STATUS_CODES = [403, 408, 429, 500, 502, 503, 504]
# Load credentials from env file
ENV_FILE = ".ods.env"
load_dotenv(dotenv_path=ENV_FILE, verbose=True)
class ODSAPIClient:
"""Example Python ODS API Client for calling the NRAO ODS API"""
def __init__(self, client_id: str, client_secret: str):
self.client_id = client_id
self.client_secret = client_secret
self.token = os.environ.get("ODS_ACCESS_TOKEN")
def refresh_token(self) -> None:
"""Fetch a new access token from the API and persist to be used for
all subsequent requests. Only re-generate tokens as necessary, not per-request.
In production, use a secure storage mechanism for client credentials.
"""
resp: requests.Response = requests.post(
TOKEN_URL,
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
},
timeout=5,
)
if resp.status_code == 200:
new_token = resp.json()["token"]
set_key(ENV_FILE, "ODS_ACCESS_TOKEN", new_token)
self.token = new_token
else:
logger.info("Failed to refresh token")
resp.raise_for_status()
def ods_request_with_retry(
self,
method: Callable,
url: str,
files: Optional[Dict[str, BufferedReader]] = None,
):
"""
Make requests to the ODS API with exponential backoff
Reuse the existing token and rotate once the ODS API indicates it is no longer valid.
"""
retries: int = 0
response: Optional[requests.Response] = None
while retries < MAX_REQUEST_RETRIES:
if not self.token:
self.refresh_token()
response = method(
url,
headers={"Authorization": f"Bearer {self.token}"},
files=files,
timeout=5,
)
if response.status_code in RETRY_STATUS_CODES:
if response.status_code == 403:
# Reactively refresh token if we get a 403
logger.info("Refreshing token")
self.refresh_token()
delay: float = backoff_delay(2, retries)
logger.info("Retrying in {} seconds...", delay)
time.sleep(delay)
retries += 1
else:
response.raise_for_status()
return response
if response is not None:
response.raise_for_status()
return response
def get_ods_data(self) -> None:
"""Example request for getting ODS Data
Client with appropriate scope required.
"""
api_response = self.ods_request_with_retry(requests.get, ODS_DATA_URL)
logger.info("Retrieved ODS data: ")
logger.info(api_response.json()["ods_data"])
def post_tle_file(self, filepath) -> None:
"""Example request for POSTing TLE files to ODS
Client with appropriate scope required.
Note, this uploads a TLE file to a live instance of NRAO ODS.
"""
with open(filepath, "rb") as f:
files = {"tle_file": f}
api_response = self.ods_request_with_retry(
requests.post, TLE_DATA_URL, files
)
logger.info("Uploaded TLE file to ODS: {}", api_response.json())
def backoff_delay(backoff_factor: int, attempts: int) -> float:
"""Add exponential backoff delay"""
return backoff_factor * (2 ** (attempts - 1))
def main(tle_file_path: Optional[str] = None) -> None:
"""
Make requests to the ODS API to get ODS Data, and upload a TLE file if applicable
"""
# Contact an NRAO ODS Admin to generate client credentials
# with applicable permissions. Store securely.
ods_client_id = os.getenv("ODS_CLIENT_ID")
ods_client_secret = os.getenv("ODS_CLIENT_SECRET")
api_client = ODSAPIClient(ods_client_id, ods_client_secret)
logger.info("Making example request to get ODS Data")
api_client.get_ods_data()
if tle_file_path:
time.sleep(2)
logger.info("Making request to POST a TLE File")
api_client.post_tle_file(tle_file_path)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="A simple Python script to demonstrate talking to the ODS API"
)
parser.add_argument(
"--tle_file_path",
type=str,
help="Optional path to a TLE file to be uploaded to ODS",
)
args = parser.parse_args()
main(args.tle_file_path)