Merge pull request #48 from Nasdaq/release-0.5.x

release-0.5.x

Author: ruchirvaninasdaq

ncdssdk/src/main/python/ncdsclient/consumer/NasdaqKafkaAvroConsumer.py

@@ -93,7 +93,7 @@ def get_kafka_consumer(self, stream_name, timestamp=None):
             if auto_offset_cfg == "earliest" or auto_offset_cfg == "smallest" or auto_offset_cfg == "beginning":
                 self.logger.debug(
                     f"Auto offset reset config set to: {auto_offset_cfg}")
-                return SeekToMidnight.seek_to_midnight_at_past_day(kafka_consumer, topic_partition, 0)
+                return SeekToMidnight.seek_to_midnight_at_past_day(kafka_consumer, topic_partition, 0, self.kafka_props.get(self.kafka_config_loader.TIMEOUT))
 
             else:
                 return kafka_consumer
@@ -105,7 +105,7 @@ def get_kafka_consumer(self, stream_name, timestamp=None):
                 self.logger.debug(
                     "offset: " + str(topic_partition.offset) + ", timestamp: " + str(timestamp))
                 offsets_for_times = kafka_consumer.offsets_for_times(
-                    [topic_partition], self.kafka_props.get(self.kafka_config_loader.TIMEOUT))
+                    [topic_partition], self.kafka_cfg.TIMEOUT)
             except Exception as e:
                 self.logger.exception(e)
                 sys.exit(0)
@@ -132,7 +132,7 @@ def get_consumer(self, avro_schema, stream_name):
             a :class:`.KafkaAvroConsumer` instance with a key and value deserializer set through the avro_schema parameter
         """
         if 'group.id' not in self.kafka_props:
-            self.kafka_props[self.kafka_config_loader.GROUP_ID_CONFIG] = f'{self.client_ID}_{stream_name}_{datetime.datetime.today().day}'
+            self.kafka_props[self.kafka_config_loader.GROUP_ID_CONFIG] = f'{self.client_ID}'
         return KafkaAvroConsumer(self.kafka_props, avro_schema)
 
     def get_schema_for_topic(self, topic):

ncdssdk/src/main/python/ncdsclient/internal/ReadSchemaTopic.py

@@ -32,8 +32,7 @@ def __init__(self):
 
     def read_schema(self, topic):
         auth_config_loader = AuthenticationConfigLoader()
-        schema_consumer = self.get_consumer(
-            "Control-" + auth_config_loader.get_client_id(self.security_props))
+        schema_consumer = self.get_consumer(auth_config_loader.get_client_id(self.security_props))
         latest_record = None
         num_messages = self.kafka_props[self.kafka_config_loader.NUM_MESSAGES]
         timeout = self.kafka_props[self.kafka_config_loader.TIMEOUT]
@@ -46,6 +45,7 @@ def read_schema(self, topic):
                 try:
                     msg_val = message.value()
 
+                    latest_record = None
                     if "name" in msg_val and msg_val["name"] == topic:
                         latest_record = message
                     if latest_record and 'schema' in msg_val:
@@ -105,7 +105,7 @@ def get_consumer(self, client_id):
             self.kafka_props = KafkaConfigLoader.load_test_config()
 
         self.kafka_props[self.kafka_config_loader.AUTO_OFFSET_RESET_CONFIG] = 'earliest'
-        self.kafka_props[self.kafka_config_loader.GROUP_ID_CONFIG] = f'{client_id}1'
+        self.kafka_props[self.kafka_config_loader.GROUP_ID_CONFIG] = f'{client_id}'
 
         kafka_avro_consumer = KafkaAvroConsumer(
             self.kafka_props, ctrl_msg_schema)
@@ -115,7 +115,7 @@ def get_consumer(self, client_id):
 
         kafka_avro_consumer.assign([topic_partition])
 
-        return SeekToMidnight.seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, 7)
+        return SeekToMidnight.seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, 6, self.kafka_props[self.kafka_config_loader.TIMEOUT])
 
     def internal_schema(self, topic):
         try:

ncdssdk/src/main/python/ncdsclient/internal/utils/SeekToMidnight.py

@@ -1,19 +1,20 @@
 import logging
 import sys
 from confluent_kafka import OFFSET_BEGINNING, OFFSET_INVALID
-from datetime import datetime, date, timedelta, time
+from datetime import datetime, timedelta
+import pytz
 
 logger = logging.getLogger(__name__)
 
 
-def seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, num_days_ago=0):
+def seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, num_days_ago=0, timeout=10):
     topic_partition.offset = get_timestamp_at_midnight(num_days_ago)
     logger.debug(
         f"Num days ago: {num_days_ago}. Setting partition offset to timestamp: {topic_partition.offset}")
     try:
         logger.debug(f"topic partition: {topic_partition}")
         offsets_for_times = kafka_avro_consumer.offsets_for_times(
-            [topic_partition], timeout=5)
+            [topic_partition], timeout=timeout)
     except Exception as e:
         logger.exception(e)
         sys.exit(0)
@@ -34,6 +35,5 @@ def seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, num_days_
 
 
 def get_timestamp_at_midnight(num_days_ago=0):
-    past_day = date.today()-timedelta(days=num_days_ago)
-    midnight = datetime.combine(past_day, time.min)
-    return int(midnight.timestamp() * 1000)
+    midnight = datetime.now(pytz.timezone('America/New_York')).replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=num_days_ago)
+    return int(midnight.timestamp() * 1000)

ncdssdk_client/src/main/python/ncdsclient/NCDSSession.py

@@ -31,7 +31,6 @@ def __init__(self, cmd):
         self.logger = logging.getLogger(__name__)
 
         self.logger = logging.getLogger(__name__)
-        self.ncds_client = None
 
     def main(self):
         self.security_cfg = load_auth_properties(self.auth_props_file)
@@ -41,42 +40,47 @@ def main(self):
         cmd_to_validate = ValidateInput(self.cmd)
         cmd_to_validate.validate_user_input()
 
+        ncds_client = None
+
         try:
-            self.ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
             if self.test_option == "TOP":
                 self.top_cmd()
 
             elif self.test_option == "SCHEMA":
+                ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
                 # Dump the Schema for the self.topic
-                schema = self.ncds_client.get_schema_for_topic(self.topic)
+                schema = ncds_client.get_schema_for_topic(self.topic)
                 print("Schema for the Topic:" + self.topic)
                 if schema:
                     print(schema)
                 else:
                     print(" Access to topic is not granted ")
 
             elif self.test_option == "GETMSG":
+                ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
                 print("Finding the message")
                 if "auto.offset.reset" in self.kafka_cfg and self.kafka_cfg["auto.offset.reset"] == "latest":
                     print("Need to get run GETMSG with 'earliest' offset")
                     sys.exit(0)
-                msg = self.ncds_client.get_sample_messages(
+                msg = ncds_client.get_sample_messages(
                     self.topic, self.message_name, False)
                 if msg is not None:
                     print(msg)
                 else:
                     print("Message Not Found ...")
 
             elif self.test_option == "GETALLMSGS":
+                ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
                 print("Finding the messages")
                 if "auto.offset.reset" in self.kafka_cfg and self.kafka_cfg["auto.offset.reset"] == "latest":
                     print("Need to run GETMSG with 'earliest' offset")
                     sys.exit(0)
-                self.ncds_client.get_sample_messages(
+                ncds_client.get_sample_messages(
                     self.topic, self.message_name, True)
 
             elif self.test_option == "TOPICS":
-                self.topics = self.ncds_client.list_topics_for_client()
+                ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
+                self.topics = ncds_client.list_topics_for_client()
                 print("List of streams available on Nasdaq Cloud Data Service:")
                 for self.topic in self.topics:
                     print(self.topic)
@@ -90,9 +94,10 @@ def main(self):
             logging.exception(e)
 
     def top_cmd(self):
+        ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
         numOfRecords = max(10, min(int(self.num_top_messages), 999))
-        records = self.ncds_client.top_messages(
-            self.topic) if not self.timestamp else self.ncds_client.top_messages(self.topic, self.timestamp)
+        records = ncds_client.top_messages(
+            self.topic) if not self.timestamp else ncds_client.top_messages(self.topic, self.timestamp)
         print("Top " + str(numOfRecords) +
               " Records for the Topic: " + self.topic)
         if records:
@@ -108,12 +113,13 @@ def top_cmd(self):
             print("Access to topic is not granted")
 
     def cont_stream_cmd(self):
-        consumer = self.ncds_client.ncds_kafka_consumer(
-            self.topic) if not self.timestamp else self.ncds_client.ncds_kafka_consumer(self.topic, self.timestamp)
+        ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
+        consumer = ncds_client.ncds_kafka_consumer(
+            self.topic) if not self.timestamp else ncds_client.ncds_kafka_consumer(self.topic, self.timestamp)
 
         try:
             while True:
-                message = consumer.poll(self.kafka_cfg[self.kafka_config_loader.TIMEOUT])
+                message = consumer.poll(self.kafka_cfg[self.kafka_config_loader.NUM_MESSAGES])
                 if message is None:
                     print(f"No Records Found for the Topic: {self.topic}")
                 else:
@@ -134,8 +140,9 @@ def filter_stream_cmd(self):
         if self.msgnames is not None:
             msgname_set = set(self.msgnames.split(","))
 
-        consumer = self.ncds_client.ncds_kafka_consumer(
-            self.topic) if not self.timestamp else self.ncds_client.ncds_kafka_consumer(self.topic, self.timestamp)
+        ncds_client = NCDSClient(self.security_cfg, self.kafka_cfg)
+        consumer = ncds_client.ncds_kafka_consumer(
+            self.topic) if not self.timestamp else ncds_client.ncds_kafka_consumer(self.topic, self.timestamp)
 
         try:
             while True:

requirements.in

@@ -4,3 +4,4 @@ oauthlib>=3.1.0,<4
 requests>=2.25.1,<3
 requests-oauthlib>=1.3.0,<2
 avro>=1.10.2,<2
+pytz>=2023.3

requirements.txt

@@ -34,3 +34,4 @@ urllib3==1.26.6
     # via requests
 zipp==3.4.1
     # via importlib-metadata
+pytz==2023.3

setup.py

@@ -1,40 +1,40 @@
-from setuptools import setup, find_packages
-import pathlib
-
-here = pathlib.Path(__file__).parent.resolve()
-
-# Get long description from README
-long_description = (here / 'README.md').read_text(encoding='utf-8')
-
-setup(
-    name='ncdssdk',
-    version='0.4.0',
-    description='A Python SDK for developing applications to access the NCDS API',
-    long_description=long_description,
-    long_description_content_type='text/markdown',
-    author='Nasdaq',
-    classifiers=[
-        # Indicate who your project is intended for
-        'Intended Audience :: Developers',
-        'Topic :: Software Development :: Build Tools',
-
-        # Pick your license as you wish
-        'License :: OSI Approved :: Apache Software License',
-
-        # Specify the Python versions you support here. In particular, ensure
-        # that you indicate you support Python 3. These classifiers are *not*
-        # checked by 'pip install'. See instead 'python_requires' below.
-        'Programming Language :: Python :: 3.9',
-    ],
-    keywords='Nasdaq, NCDS, ncdssdk',
-    packages=find_packages(),
-    python_requires='>=3.9, <4',
-    install_requires=open("requirements.in").readlines(),
-
-    include_package_data=True,
-    # We could possibly use entry_points parameter with console_scripts here to specify the NCDSSession script
-
-    project_urls={  # Optional
-        'Source': 'https://github.com/Nasdaq/NasdaqCloudDataService-SDK-Python',
-    },
-)
+from setuptools import setup, find_packages
+import pathlib
+
+here = pathlib.Path(__file__).parent.resolve()
+
+# Get long description from README
+long_description = (here / 'README.md').read_text(encoding='utf-8')
+
+setup(
+    name='ncdssdk',
+    version='0.5.0',
+    description='A Python SDK for developing applications to access the NCDS API',
+    long_description=long_description,
+    long_description_content_type='text/markdown',
+    author='Nasdaq',
+    classifiers=[
+        # Indicate who your project is intended for
+        'Intended Audience :: Developers',
+        'Topic :: Software Development :: Build Tools',
+
+        # Pick your license as you wish
+        'License :: OSI Approved :: Apache Software License',
+
+        # Specify the Python versions you support here. In particular, ensure
+        # that you indicate you support Python 3. These classifiers are *not*
+        # checked by 'pip install'. See instead 'python_requires' below.
+        'Programming Language :: Python :: 3.9',
+    ],
+    keywords='Nasdaq, NCDS, ncdssdk',
+    packages=find_packages(),
+    python_requires='>=3.9, <4',
+    install_requires=open("requirements.in").readlines(),
+
+    include_package_data=True,
+    # We could possibly use entry_points parameter with console_scripts here to specify the NCDSSession script
+
+    project_urls={  # Optional
+        'Source': 'https://github.com/Nasdaq/NasdaqCloudDataService-SDK-Python',
+    },
+)