How to use File Notification Message Distribution with the Client File Store (CFS) Service

January 31,2024

Wasin Waeosri

Overview

Last Update April 2024

Important for Messenger Compliance External feeds users: On 28 April 2024, a new version of Messenger will be released in parallel with a new External Feeds offering, which will allow customers to retrieve their compliance files in JSON format using this Client File Store (CFS) API. For more details on support, refer to PCN208738.

The Data Platform Client File Store (CFS) Service offers an alternative way to be notified when the new FileSet available through Amazon Simple Queue Service (SQS). The consumers do not need to polling request to fileset API, but continuously monitor the queue and received all required information (including file id) and able to download new file directly through file stream api instead.

This project is the part 2 of the A Step-By-Step Workflow Guide for RDP Client File Store (CFS) API project. This project describes how to use the Data Platform's Message services delivery API to notified consumers (aka subscribers) when the new bulk file is available on the CFS API, and demonstrate with the ready-to-use example tool.

What is Refinitiv Data Platform (RDP) APIs?

Let’s start with what the Refinitiv Data Platform APIs are. The Refinitiv Data Platform (RDP) APIs provide various Refinitiv data and content for developers via easy-to-use Web-based API.

RDP APIs give developers seamless and holistic access to all of the Refinitiv content such as Environmental Social and Governance (ESG), News, Research, etc, and commingled with their content, enriching, integrating, and distributing the data through a single interface, delivered wherever they need it. The RDP APIs delivery mechanisms are the following:

Request - Response: RESTful web service (HTTP GET, POST, PUT or DELETE)
Alert: delivery is a mechanism to receive asynchronous updates (alerts) to a subscription.
Bulks: deliver substantial payloads, like the end-of-day pricing data for the whole venue.
Streaming: deliver real-time delivery of messages.

This example project is focusing on the Request-Response: RESTful web service delivery method only.

Request - Response: RESTful web service (HTTP GET, POST, PUT or DELETE)
Alert: delivery is a mechanism to receive asynchronous updates (alerts) to a subscription.
Bulks: deliver substantial payloads, like the end-of-day pricing data for the whole venue.
Streaming: deliver real-time delivery of messages.

This example project is focusing on the Request-Response: RESTful web service delivery method only.

For more detail regarding the Refinitiv Data Platform, please see the following APIs resources:

Quick Start page.
Tutorials page.

That covers the overview of the RDP APIs.

What is CFS?

My next point is CFS service. The Client File Store (CFS) aka File Distribution is a capability of Refinitiv Data Platform (RDP) that provides authorization and enables access to content files stored in publisher-supplied repository. CFS defines content ownership that publisher are isolated. And subscribers can trust the source of content.

CFS is engineered as a self-service metadata tool intend for publishers and subscribers. CFS provides bucket and file-set to organize files to simplify the interaction with publishers or subscribers CFS doesn't store file directly. Actual files are store in publisher-supplied. AWS S3 only one type storage that supported by current CFS.

That’s all I have to say about the CFS service.

What is Message Distribution Service?

So, now let’s look at the Message Distribution Service. The Message Distribution Service is a service that allows customers to subscribe to message queue-based interfaces to retrieve non-real-time content changes for all datasets. The Refinitiv Data Platform currently utilizes AWS SQS as a main message queue technology.

Integrated with Client File Store (CFS) Service, message distribution will deliver the message notification for end user clients when the client's interested bulk file is available.

Note:

The equivalent message queue service on Microsoft Azure is Queue Storage service.
The most equivalent message queue service on Google Cloud Platform is Pub/Sub service.

How the File Notification Message Distribution work?

That brings us to a technical side of the File Notification Message Distribution service. The service lets the clients subscribes for the notification of their interested CFS file (using Bucket names, Package Id, etc) to the RDP Message Distribution Service. Then the clients continuously monitor the queue and received all required information, and able to download new file directly through the RDP file stream api when the file is available.

File Notification Message Distribution diagram

File Notification Message Distribution Workflow Step-By-Step

The consumer (subscriber) application needs to follow though the activities below in order to receive file notification and download CFS file.

Subscribers make a subscription to API https://api.refinitiv.com/message-services/v1/file-store/subscriptions endpoint by specified CFS content that they interest. E.g. Bucket Name, Package Id, Fileset Attribute.
Subscribers received
- Subscription ID
- SQS Endpoint
- Decryption Key
Subscribers request credential to access the provided SQS.
Subscribers continually checking the SQS in short interval and as soon as the new FileSet that matched with the criteria available message will be available in user specific SQS
Fetch message from SQS and decrypt the message to see FileSet / File detail
Call CFS API endpoint to generate the pre-signed URL to download the file.

File Notification Message Distribution Tools

That brings us to the File Notification Message Distribution Tools which do all the above steps and help you download the file to the specified location automatically.

The Tools are based on full-function Python scripts and require the following prerequisite, dependencies as follows:

Prerequisite

Access to the RDP with the your desire CFS file permission

The tools uses RDP access credentials with the CFS file permission. You need both RDP User-ID (email base) and Machine-ID (GE-A-XXXXX) user types.

User-ID: Identifier for user allowing access to contracted content and APIs on the API Playground page (https://apidocs.refinitiv.com/Apps/ApiDocs) or Bulk UI via Workspace/Eikon Desktop. It normally is email address of user (example: sample@lseg.com).
Machine ID: Identifier for machine allowing access and run this tool or any automate applications. It normally is a username with "GE-A-XXXXXXXX" format.

Please see more detail about how to setup your RDP User-ID and Machine-ID from the Getting Started with Refinitiv Data Platform article.

You can contact your LSEG representative to help you with the RDP account and services.

Internet Access

The tools need to download the required libraries from the https://pypi.org/ Python package repository website. The CFS file is also available on AWS Cloud, so you need internat access to use the tool.

Python

The tools uses Python runtime. The Python Anaconda or MiniConda distribution/package manager is also supported.

How to run the Tools

The first step is to unzip or download the example project folder into a directory of your choice, then set up Python environment based on your preference.

You can download the tool via the following Git command or manual download from the page.

Download via Git

Firstly, you need to install Git tool on your machine. You can find more detail about how to install Git from the following resources:

Once the installation is succeed, please open a Git Bash tool on the prefer folder and run the following command:

    	
            git clone https://github.com/LSEG-API-Samples/FileNoti.Message.Distribution.Tools.git

Manual Download

Please open the https://github.com/LSEG-API-Samples/FileNoti.Message.Distribution.Tools/tree/main URL on your prefer web browser, click on the "Code" button and choose "Download ZIP" option.

Tool Setup with Python

Firstly, if you didn't have python3.7 or python 3.9 yet please install it via https://www.python.org/downloads/ website.

You can follow https://wiki.python.org/moin/BeginnersGuide/Download and https://realpython.com/installing-python/ websites for the installation guides.

Once the installation process succeeded, please verify python version before you proceed the next step (Readme file currently support installation python3.7 and python3.9)

    	
            python --version

Next, run the followings commands to install python libraries

// For python3.7

    	
            python3 -m pip install -r python37_libs.txt

// For python 3.9

    	
            python3 -m pip install -r python39_libs.txt

If you encounter error ModuleNotFoundError: No module named 'Crypto' please follow step below // This step is used for python3.7 troubleshooting.

    	
            python -m pip uninstall crypto 

python -m pip uninstall pycrypto

python -m pip install -r python37_libs.txt

Now we are ready to setup the RDP credential. Go to folder name msg_dist_tools and open file credentials.ini and specify your RDP Machine-ID credential type information as follows (If you don't know information please contact your LSEG representative).

    	
            [RDP]

username = <RDP Machine-ID GE-A-XXXXXXXX>

password = <RDP Machine-ID password>

clientId = <RDP App Key>

Caution: Please do not set your RDP User-ID (email) credential type or the same user to access Workspace here.

To Run Program please check Tool Description section. The Messages will be stored under metadata/ folder and the FileNoti file will be downloaded into your destination folder.

Tool Setup with Anaconda/MiniConda

Firstly, Download Python Anaconda or MiniConda distribution/package manager and install it in your local machine.

Then Open Anaconda Prompt and go to the project's folder and run the following command in the Anaconda Prompt application to create a Conda environment named CFS_FileNoti for the project.

    	
            (base) $>conda create --name CFS_FileNoti python=3.9

Once the environment is created, activate a Conda CFS_FileNoti environment with this command in Anaconda Prompt.

    	
            (base) $>conda activate CFS_FileNoti

Then run the following command to the dependencies in the CFS_FileNoti environment.

    	
            (CFS_FileNoti) $>pip install -r python39_libs.txt

    	
            [RDP]

username = <RDP Machine-ID GE-A-XXXXXXXX>

password = <RDP Machine-ID password>

clientId = <RDP App Key>

Caution: Please do not set your RDP User-ID (email) credential type or the same user to access Workspace here.

To Run Program please check Tool Description section. The Messages will be stored under metadata/ folder and the FileNoti file will be downloaded into your destination folder.

Tools Running Description

My next point is how to run the tool. Firstly, you need to set Bucket-name and Package Id (if you have the package Id information) in the json file in the msg_dist_tools/requestBody folder.

Example: singleBucketFilter.json

    	
            {
    "transport": {
        "transportType": "AWS-SQS"
    },
    "query": {
        "bool": {
            "must": [
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.bucketName": "{bucket-name}"
                    }
                },
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.status": "READY"
                    }
                }
            ]
        }
    }
}

Example: singlePackageIdFilter.json

    	
            {
    "transport": {
        "transportType": "AWS-SQS"
    },
    "query": {
        "bool": {
            "must": [
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.packageId": "{package id}"
                    }
                },
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.status": "READY"
                    }
                }
            ]
        }
    }
}

Example: bucketAndPackageIdFilter.json

    	
            {
    "transport": {
        "transportType": "AWS-SQS"
    },
    "query": {
        "bool": {
            "must": [
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.bucketName": "{bucket-name}"
                    }
                },
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.packageId": "{package id}"
                    }
                },
                {
                    "term": {
                        "payload.FileStoreNotification.fileset.status": "READY"
                    }
                }
            ]
        }
    }
}

The examples of Bucket-names and Package IDs are as follows:

Content	Bucket Name	Example of Package ID
Financial Markets Reference Data	bulk-FMRD	4d48-d7ff-edcc3d38-8243-a4f7517962b8
Symbology	bulk-Symbology	4c80-73a0-fcef949b-bfde-2b9b8117cfb0
ESG	bulk-ESG	4288-ebb6-93372235-acb2-89882a826af1
ESG - Point in Time	bulk-ESG	4173-aec7-8a0b0ac9-96f9-48e83ddbd2ad
Tick History	TICKHISTORY_VBD_NO_EMBARGO	4c01-ab9e-db594a31-a8f5-5b7852ec4638
Green Revenue	bulk-GreenRevenue	Summary: 4e94-6d63-fea034dc-90e2-de33895bd4e9
Green Revenue	bulk-GreenRevenue	Standard: 4316-d43b-81c40763-8e6a-0dbec8162ab1
Starmine	STARMINE_PREDICTIVE_ANALYTICS_SMARTECON_LIVE	40d4-1404-58533484-afe8-718650a4e0d4
Municipal Market Monitor (TM3)	bulk-Custom	4c5b-0f65-c57f5ae4-b548-2a1075ee725d

Please contact your LSEG representative about your Bucket-name and Package ID.

The next step is creating a new subscription and specify input json file with the following command:

    	
            python fileNotiMessages.py -c -i requestBody/<json file>

Example: I am demonstrating with singleBucketFilter.json and Test_CFS_BucketTiming test bucket-name and 4fbb-f1b5-d428da4f-9392-28214215d35a test package Id as an example data set.

    	
            python fileNotiMessages.py -c -i requestBody/bucketAndPackageIdFilter.json

Result:

    	
            log_path log\app.log, log_dir log
log_path log\error.log, log_dir log
Program is started
\ 2024-01-18 18:04:20,861 INFO Successfully get current user: {GE-A-XXXXXXXX}
2024-01-18 18:04:20,861 INFO input parameter = {'get': False, 'create': True, 'poll': False, 'modify': False, 'delete': False, 'subscriptionId': None, 'input': 'requestBody/bucketAndPackageIdFilter.json', 'destinationFolder': None, 'queue': False, 'recovery': False}
2024-01-18 18:04:20,861 INFO
 
2024-01-18 18:04:20,861 INFO *************************************************************************
2024-01-18 18:04:20,861 INFO ******************* Create new subscription *****************************
2024-01-18 18:04:20,861 INFO *************************************************************************
2024-01-18 18:04:20,861 INFO Request URL: https://api.refinitiv.com/message-services/v1/file-store/subscriptions
2024-01-18 18:04:20,877 INFO Request Body: {'transport': {'transportType': 'AWS-SQS'}, 'query': {'bool': {'must': [{'term': {'payload.FileStoreNotification.fileset.bucketName': 'Test_CFS_BucketTiming'}}, {'term': {'payload.FileStoreNotification.fileset.packageId': '4000-04ea-7d282653-8e6d-810dda18956d'}}, {'term': {'payload.FileStoreNotification.fileset.status': 'READY'}}]}}}
/ 2024-01-18 18:04:22,912 INFO -------------------- Successfully create subscription ----------------------
2024-01-18 18:04:22,928 INFO subscriptionID: b6447a1a-da72-4fbd-b2f5-96185d72eb21
2024-01-18 18:04:22,928 INFO transportEndpoint: https://sqs.us-east-1.amazonaws.com/642157181326/sqs-edsalerts-main-prod-usersqs-b6447a1a-da72-4fbd-b2f5-96185d72eb21
2024-01-18 18:04:22,928 INFO cryptographyKey: iT4kQg+7eQnYhphLN8FbxrfeTXHLGKo0dbHapGm32pQ=
2024-01-18 18:04:22,928 INFO query: {'bool': {'must': [{'term': {'payload.FileStoreNotification.fileset.bucketName': 'Test_CFS_BucketTiming'}}, {'term': {'payload.FileStoreNotification.fileset.packageId': '4000-04ea-7d282653-8e6d-810dda18956d'}}, {'term': {'payload.FileStoreNotification.fileset.status': 'READY'}}]}}

The information you need is the subscriptionID in the log above. Please copy it for further use.

The next step is polling message queue from existing subscription and specify destination folder using the subscriptionID above.

    	
            python fileNotiMessages.py -p -s <subscriptionId> -d <destination folder>

Example:

    	
            python fileNotiMessages.py -p -s b6447a1a-da72-4fbd-b2f5-96185d72eb21 -d C:\msg_dist_python_tools\cfs_download

If you want to change the Bucket-name or Package Id, you can update the current subscriptionID with the updated JSON file like the following example:

    	
            python fileNotiMessages.py -m -s b6447a1a-da72-4fbd-b2f5-96185d72eb21 -i requestBody/multipleBucketFilter.json

Caution: If you still want a CFS file, please keep the subscriptionID (and update it if you want to change something) until you really want to stop download the CFS file.

If you want to stop download the CFS file, you can delete the subscriptions with the following command:

    	
            python fileNotiMessages.py -u

List of supported commands

Please check the list of all supported commands from the GitHub repository.

That covers how to run the tools.

Tools Troubleshooting

Moving on to troubleshooting the problems that you may encounter when setup or use the tools.

Error: "'pip' is not recognized as an internal or external command, operable program or batch file." when run a "python3 -m pip install" command.

Answer: Your Python installation is not completed. You need to add the path of the pip installation ({Python installation folder}\Scripts folder by default) to your PATH system variable. For more detail, please check the following resources:

Error: "Failed to get access token 400 - {"error":"access_denied" ,"error_description":"Invalid username or password." }" when running the tool.

Answer: This error message means your RDP username or password information in the msg_dist_tools/credentials.ini is invalid. Please verify your RDP username (user-id/machine-id) and password.

Error: "Failed to get access token 401 - {"error":"invalid_client" ,"error_description":"Invalid Application Credential." }" when running the tool.

Answer: This error message means your RDP App Key information in the msg_dist_tools/credentials.ini is invalid. Please verify your RDP App Key.

Error: Error with "{"status":403,"message":"Access denied: Insufficient claims to access this resource"}" when running the tool.

Answer: This error message means your RDP account does not have permission to access the RDP CFS API for your desire bucket. Please contact your LSEG representative to verify your permission.

Conclusion

Before I finish, let me just say The Message Distribution service is a powerful service that allows consumers to receive content changes notification of non-real-time content from the message queue-based interfaces. Powered by Amazon SQS (as of January 2024), developers can create a consumer application to connect to the Service with various programming languages using AWS SDK. Combined with Client File Store (CFS) service, the File Notification Message Distribution service lets consumers receive notification when the interested CFS file is available and be able to download that file without manually tasks.

At the same time, the Refinitiv Data Platform (RDP) APIs provide various LSEG data and content for developers via an easy-to-use Web-based API. The APIs are easy to integrate into any application and platform that supports the HTTP protocol and JSON message format.

References

That brings me to the end of my File Notification Message Distribution with the CFS Service project. For further details, please check out the following resources: