Using DEDL API Keys for Automated Workflows

import os
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from tqdm import tqdm
import json
from getpass import getpass
from datetime import datetime, timedelta, timezone

from destinelab import DEDLServiceAccountAuth

# API key provisioned for a Service Account
HDA_CLIENT_ID = os.environ.get("HDA_CLIENT_ID")
HDA_CLIENT_SECRET = os.environ.get("HDA_CLIENT_SECRET")

if (HDA_CLIENT_ID is None or HDA_CLIENT_SECRET is None):
        HDA_CLIENT_ID = input("Please input your client ID: ")
        HDA_CLIENT_SECRET = getpass("Please input your API key: ")
        os.environ["HDA_CLIENT_ID"] = HDA_CLIENT_ID
        os.environ["HDA_CLIENT_SECRET"] = HDA_CLIENT_SECRET

def requestANewToken():
    try:
        access_token = DEDLServiceAccountAuth(client_id=HDA_CLIENT_ID,client_secret=HDA_CLIENT_SECRET).get_token()
    
        if access_token is not None:
            print("DEDL/DESP Access Token Obtained Successfully")
    except Exception as e:
        print("Failed to Obtain DEDL/DESP Access Token", e)

    return {"Authorization": f"Bearer {access_token}"}

auth_headers = requestANewToken()

COLLECTION_ID = "EO.MO.DAT.OCEANCOLOUR_GLO_BGC_L3_NRT_009_101"

# Current time in UTC
now_utc = datetime.now(timezone.utc)
# Yesterday's date (UTC)
one_week_ago = (now_utc - timedelta(days=7)).date()

# Construct full UTC day interval
start = datetime.combine(one_week_ago, datetime.min.time(), tzinfo=timezone.utc)
end = datetime.combine(now_utc, datetime.max.time(), tzinfo=timezone.utc)

datetime_range = f"{start.isoformat().replace('+00:00', 'Z')}/{end.isoformat().replace('+00:00', 'Z')}"

request_body = {
    "collections": [
        COLLECTION_ID,
    ],
    "datetime": datetime_range
}
request_body

BASE_URL = "https://hda.data.destination-earth.eu/stac/v2" 

response=requests.post(BASE_URL+'/search', json=request_body, headers=auth_headers)

response.raise_for_status()

datasets=[] 
for i in response.json().get("features"):
    datasets.append({"id":i.get("id"),"downloadLink": i.get("assets").get("downloadLink").get("href"),\
                    # "alternate": i.get("assets").get("downloadLink").get("alternate").get("origin").get("href")
                    "datetime": i.get("properties").get("datetime")
                    })
from pprint import pprint
pprint(datasets)

#Sometimes requests to polytope get timeouts, it is then convenient define a retry strategy
retry_strategy = Retry(
    total=5,  # Total number of retries
    status_forcelist=[500, 502, 503, 504],  # List of 5xx status codes to retry on
    allowed_methods=["GET",'POST'],  # Methods to retry
    backoff_factor=1  # Wait time between retries (exponential backoff)
)

# Create an adapter with the retry strategy
adapter = HTTPAdapter(max_retries=retry_strategy)

# Create a session and mount the adapter
session = requests.Session()
session.mount("https://", adapter)

def download(href,auth_headers=auth_headers):

    response = session.get(href, stream=True, headers=auth_headers)
    
    if response.status_code == 401:
        auth_headers = requestANewToken()
        response = session.get(href, stream=True, headers=auth_headers)

    response.raise_for_status()
    
    content_disposition = response.headers.get('Content-Disposition')
    total_size = int(response.headers.get("content-length", 0))
    if content_disposition:
        filename = content_disposition.split('filename=')[1].split('"')[1]
    else:
        filename = os.path.basename(url)
        
    # Open a local file in binary write mode and write the content
    print(f"downloading {filename}")
    
    with tqdm(total=total_size, unit="B", unit_scale=True) as progress_bar:
        with open(filename, 'wb') as f:
            for data in response.iter_content(1024):
                progress_bar.update(len(data))
                f.write(data)
    
    result=filename+' NOT'
    if (os.path.exists(filename)):
        result=filename
    return result

#select only plankton data at 300mt
target = "plankton_nrt_l3-olci-300m_P1D"

filtered = [
    d for d in datasets
    if target in d.get("id", "") and "datetime" in d
]

if not filtered:
    raise RuntimeError("No datasets found matching the requested product")

#select only the most recent data
most_recent = max(
    filtered,
    key=lambda d: datetime.fromisoformat(d["datetime"].replace("Z", "+00:00"))
)

download(most_recent["downloadLink"])