Example Python Script

An example Python script for reference only.

"""
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)