Downloading DSS data via Python API

Hi all,

I'm am currently trying to use DSS via Python API. After running the code, I see that I am able to create a new instrument list, report template and schedule via my Python. However, I am than not able to retrieve the document with the data.

This is the code that I use for downloading the data:

while True:        

    response = requests.get(files_url, headers=headers)        

    response.raise_for_status()        

    files = response.json()        

    partial_files = [f for f in files['value'] if f['FileType'] == 'Partial']        

    if partial_files:            

        for file in partial_files:                

            file_id = file['ExtractedFileId']                

            file_name = file['ExtractedFileName']                

            file_url = f"{EXTRACTED_FILE_URL}('{file_id}')/$value"                

            file_resp = requests.get(file_url, headers=headers)                

            with open(file_name, 'wb') as f:                    

                f.write(file_resp.content)                

            print(f"Saved partial data file: {file_name}")            

        break        

    else:            

        print("Waiting for partial data files...")            

        time.sleep(10)

I get the following error after awhile:



requests.exceptions.ConnectionError:

HTTPSConnectionPool(host='selectapi.datascope.refinitiv.com', port=443):

Max retries exceeded with url: ...

Caused by NameResolutionError: getaddrinfo failed

Does anyone know how to fix this issue? Many thanks in advance.

Find more posts tagged with

python

dss-rest-api

#technology

Accepted answers

All comments

Jirapongse

@TomCroese

Thank you for reaching out to us.

Typically, this error indicates that the machine can't resolve the host name. For example, if I use an invalid URL with the requests.get method, I will get the following error.

ConnectionError: HTTPSConnectionPool(host='www.wrongurlxxx.com', port=443): Max retries exceeded with url: / (Caused by NameResolutionError("<urllib3.connection.HTTPSConnection object at 0x00000281920378C0>: Failed to resolve 'www.wrongurlxxx.com' ([Errno 11001] getaddrinfo failed)"))

I can run the following code properly.

import requests

response = requests.get("https://selectapi.datascope.refinitiv.com")

response.status_code

This may relate to the networking settings on your machine.

TomCroese

Thank you for your answer.

So far my used URL seems to be correct:

DSS_URL = 'https://selectapi.datascope.refinitiv.com/RestApi/v1/'

This URL is also used in the functions for getting the token, creating the instrument list, etc. Those parts of the code run without any connection issues.

Do I need to use a different URL for the downloading? Or is there another way to solve this issue?

Thanks in advance.

Jirapongse

@TomCroese

Yes, the URL is correct but I don't why Python can't resolve it.

The files_url should look like this:

https://selectapi.datascope.refinitiv.com/RestApi/v1/Extractions/ReportExtractions('548722918')/Files

The file_url should look like this:

https://selectapi.datascope.refinitiv.com/RestApi/v1/Extractions/ExtractedFiles('VjF8fDkwMjQzODM5NQ')/$value

You can refer to the article and sample Python code.

Please share the full code so I can test it on my machine.

TomCroese

@Jirapongse

Thank you, I will have a look at the article and the sample code.

In the mean time, this is the current code I am using:

import requests

import time

import gzip

import os

from datetime import datetime, timedelta, timezone


# === USER CONFIGURATION ===


DSS_URL = 'https://selectapi.datascope.refinitiv.com/RestApi/v1/'

TOKEN_URL = DSS_URL + 'Authentication/RequestToken'

EXTRACT_WITH_NOTES_URL = DSS_URL + 'Extractions/ExtractWithNotes'

INSTRUMENT_LIST_URL = DSS_URL + 'Extractions/InstrumentLists'

INTRADAY_PRICING_REPORT_TEMPLATE_URL = DSS_URL + 'Extractions/IntradayPricingReportTemplates'

SCHEDULE_URL = DSS_URL + 'Extractions/Schedules'

REPORT_EXTRACTION_URL = DSS_URL + 'Extractions/ReportExtractions'

EXTRACTED_FILE_URL = DSS_URL + 'Extractions/ExtractedFiles'

REPORT_TEMPLATE_URL = DSS_URL + 'Extractions/ReportTemplates'


# Fill your credentials here

USERNAME = '...'

PASSWORD = '...'


# Your instrument list example (modify to your RICs)

INSTRUMENTS = [

    {'IdentifierType': 'Ric', 'Identifier': 'TFMBYc1'}

]


INSTRUMENT_LIST_NAME = "MyInstrumentList"

REPORT_TEMPLATE_NAME = "MyIntradayPricingTemplate"

SCHEDULE_NAME = "MySchedule"


TOKEN = None


def get_token():

    global TOKEN

    payload = {

        "Credentials": {

            "Username": USERNAME,

            "Password": PASSWORD

        }

    }

    headers = {'Prefer': 'respond-async'}

    response = requests.post(TOKEN_URL, json=payload, headers=headers)

    response.raise_for_status()

    data = response.json()

    TOKEN = data.get('value')

    if not TOKEN:

        raise Exception("Failed to retrieve token")

    print("Token retrieved")

    return TOKEN


def headers_json():

    return {

        'Authorization': f'Token {TOKEN}',

        'Content-Type': 'application/json',

        'Prefer': 'respond-async'

    }


def create_instrument_list():

    payload = {

        "@odata.type": "#DataScope.Select.Api.Extractions.SubjectLists.InstrumentList",

        "Name": INSTRUMENT_LIST_NAME

    }

    response = requests.post(INSTRUMENT_LIST_URL, json=payload, headers=headers_json())

    response.raise_for_status()

    data = response.json()

    list_id = data["ListId"]

    print(f"Instrument list created with ID: {list_id}")


    # Append instruments

    append_url = f"{INSTRUMENT_LIST_URL}('{list_id}')/DataScope.Select.Api.Extractions.InstrumentListAppendIdentifiers"

    append_payload = {

        "Identifiers": INSTRUMENTS,

        "KeepDuplicates": False

    }

    append_resp = requests.post(append_url, json=append_payload, headers=headers_json())

    append_resp.raise_for_status()

    appended = append_resp.json()

    print(f"Appended {appended['AppendResult']['AppendedInstrumentCount']} instruments")

    return list_id


def create_report_template():

    content_fields = [{"FieldName": f} for f in [

        'RIC', 'Last Price', 'Last Update Time', 'Last Volume', 'Bid Price', 'Ask Price', 'Bid Size', 'Ask Size'

    ]]

    payload = {

        "@odata.type": "#DataScope.Select.Api.Extractions.ReportTemplates.IntradayPricingReportTemplate",

        "ShowColumnHeaders": True,

        "Name": REPORT_TEMPLATE_NAME,

        "Headers": [],

        "Trailers": [],

        "ContentFields": content_fields,

        "Condition": {}

    }

    response = requests.post(INTRADAY_PRICING_REPORT_TEMPLATE_URL, json=payload, headers=headers_json())

    response.raise_for_status()

    data = response.json()

    print(f"Report template created with ID: {data['ReportTemplateId']}")

    return data['ReportTemplateId']


def schedule_extraction(instrument_list_id, report_template_id, start_time, end_time):

    # You may need to adjust the schedule payload to specify time window if API supports it.

    # For this example, scheduling immediate extraction limited to today's data.

    # You might want to add a 'Condition' for time window if API accepts.


    payload = {

        "Name": SCHEDULE_NAME,

        "Recurrence": {

            "@odata.type": "#DataScope.Select.Api.Extractions.Schedules.SingleRecurrence",

            "IsImmediate": True

        },

        "Trigger": {

            "@odata.type": "#DataScope.Select.Api.Extractions.Schedules.ImmediateTrigger",

            "LimitReportToTodaysData": True

        },

        "ListId": instrument_list_id,

        "ReportTemplateId": report_template_id

    }

    response = requests.post(SCHEDULE_URL, json=payload, headers=headers_json())

    response.raise_for_status()

    data = response.json()

    schedule_id = data['ScheduleId']

    print(f"Extraction scheduled with ID: {schedule_id}")

    return schedule_id


def wait_for_extraction(schedule_id):

    pending_url = f"{SCHEDULE_URL}('{schedule_id}')/PendingExtractions"

    completed_url = f"{SCHEDULE_URL}('{schedule_id}')/CompletedExtractions"

    headers = {'Authorization': f'Token {TOKEN}'}


    print("Waiting for extraction to complete...")


    while True:

        response = requests.get(pending_url, headers=headers)

        response.raise_for_status()

        pending = response.json()

        if pending["value"]:

            print("Extraction still pending...")

            time.sleep(10)

        else:

            response = requests.get(completed_url, headers=headers)

            response.raise_for_status()

            completed = response.json()

            if completed["value"]:

                extraction_id = completed["value"][0]["ReportExtractionId"]

                print(f"Extraction completed with ID: {extraction_id}")

                return extraction_id

            else:

                print("No extraction found yet, retrying...")

                time.sleep(10)


def download_data(extraction_id, filename):

    notes_url = f"{REPORT_EXTRACTION_URL}('{extraction_id}')/NotesFile"

    files_url = f"{REPORT_EXTRACTION_URL}('{extraction_id}')/Files"

    headers = {'Authorization': f'Token {TOKEN}'}


    # Wait for notes file availability

    while True:

        response = requests.get(notes_url, headers=headers)

        if response.status_code == 200:

            notes_info = response.json()

            break

        else:

            print("Waiting for notes file...")

            time.sleep(5)


    # Download notes

    notes_file_id = notes_info["ExtractedFileId"]

    notes_file_name = notes_info["ExtractedFileName"]

    notes_file_url = f"{EXTRACTED_FILE_URL}('{notes_file_id}')/$value"

    notes_resp = requests.get(notes_file_url, headers=headers)

    with open(notes_file_name, 'wb') as f:

        f.write(notes_resp.content)

    print(f"Saved notes file: {notes_file_name}")


    # Wait for partial data files and download them

    while True:

        response = requests.get(files_url, headers=headers)

        response.raise_for_status()

        files = response.json()

        partial_files = [f for f in files['value'] if f['FileType'] == 'Partial']

        if partial_files:

            for file in partial_files:

                file_id = file['ExtractedFileId']

                file_name = file['ExtractedFileName']

                file_url = f"{EXTRACTED_FILE_URL}('{file_id}')/$value"

                file_resp = requests.get(file_url, headers=headers)

                with open(file_name, 'wb') as f:

                    f.write(file_resp.content)

                print(f"Saved partial data file: {file_name}")

            break

        else:

            print("Waiting for partial data files...")

            time.sleep(10)


def main():

    get_token()


    instrument_list_id = create_instrument_list()

    report_template_id = create_report_template()


    # For your 2-hour window yesterday, get times in ISO format if API supports time condition (example only)

    yesterday = datetime.utcnow().date() - timedelta(days=1)

    start_time = datetime.combine(yesterday, datetime.min.time()) + timedelta(hours=10)  # 10:00 UTC yesterday

    end_time = start_time + timedelta(hours=2)  # 2 hour window


    print(f"Scheduling extraction for {start_time} to {end_time} UTC (Note: Adjust if API supports)")


    schedule_id = schedule_extraction(instrument_list_id, report_template_id, start_time, end_time)


    extraction_id = wait_for_extraction(schedule_id)


    download_data(extraction_id, "output.csv")


if __name__ == "__main__":

    main()

Jirapongse

@TomCroese

Thank you for the code.

I can run it properly on my machine so the code is correct.

Token retrieved

Instrument list created with ID: 0x099806a5a00b6bed

Appended 1 instruments

Report template created with ID: 0x0997eea40ebb6bd7

Scheduling extraction for 2025-10-22 10:00:00 to 2025-10-22 12:00:00 UTC (Note: Adjust if API supports)

C:\Users\jphuriph\AppData\Local\Temp\ipykernel_20972\567012375.py:382: DeprecationWarning: datetime.datetime.utcnow() is deprecated and scheduled for removal in a future version. Use timezone-aware objects to represent datetimes in UTC: datetime.datetime.now(datetime.UTC).

  yesterday = datetime.utcnow().date() - timedelta(days=1)

Extraction scheduled with ID: 0x09983980a0ab6c25

Waiting for extraction to complete...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

…

…

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction still pending...

Extraction completed with ID: 2000001027796676

Saved notes file: 9008895.MySchedule.20251023.151105.2000001027796676.x10q02.csv.notes.txt

Saved partial data file: 9008895.MySchedule.20251023.151105.2000001027796676.x10q02.10min.csv

Saved partial data file: 9008895.MySchedule.20251023.151105.2000001027796676.x10q02.0min.csv

The problem should not be in the code. It could be a network issue or network settings.

TomCroese

Hi @Jirapongse

Thank you for testing my full code. Good to know that the code itself is not the issue.

When running the code, I am also able to do the extraction. It then saves the notes file as txt, so that also works correctly. Then I get stuck with the "Waiting for partial data files…". So this part is still not working.

In the saved notes file, we see the embargo on the file:

The file ….MySchedule_test.20251027.104707.2000001030503229.x09q30.csv will be embargoed until 27/10/2025 10:57:08.

27/10/2025 10:57:08: Embargoed file ….MySchedule_test.20251027.104707.2000001030503229.x09q30.csv released.

However, the code is still running after the embargo time, and not able to download the csv.

How can I best proceed from here? How could I try to fix this issue?

Many thanks in advance.

Jirapongse

@TomCroese

You can check this How to Retrieve Early Partial Delivery of Embargoed Reports via DataScope Select REST API article.

EXPLORE OUR SITES