Merge pull request #5 from bytesview/csv_update

added feature to save csv file

Author: arjunjain

newsdataapi/helpers.py

@@ -1,4 +1,4 @@
-import time
+import time,os,csv
 
 def get(request_method, URL, URL_parameters_encoded, proxies, request_timeout):
     if proxies is None:
@@ -13,4 +13,44 @@ def MaxRetries(response, max_retries, retry_delay, request_method, URL, URL_para
         if response.status_code!=500:
             break
         max_retries-=1
-    return response
+    return response
+
+
+class FileHandler:
+
+    def __init__(self,folder_path:str=None) -> None:
+        self.folder_path = folder_path
+
+    def generate_csv_file(self,collected_data:list,full_filepath:str):
+        with open(full_filepath,'w',newline='', encoding='utf-8') as csv_file:
+            writer = csv.DictWriter(csv_file,fieldnames=list(collected_data[0].keys()))
+            writer.writeheader()
+            writer.writerows(collected_data)
+
+
+    def save_to_csv(self, response: dict, folder_path: str=None, filename: str=None)->str:
+        folder_path = folder_path or self.folder_path
+        filename = filename or str(time.time_ns())
+
+        if not folder_path or not os.path.exists(folder_path):
+            raise FileNotFoundError(f'Provided folder path not found: {folder_path}')
+
+        filename = f'{filename}.csv' if not filename.endswith('.csv') else filename
+        full_path = os.path.join(folder_path, filename)
+
+        if os.path.exists(full_path):
+            raise FileExistsError(f'Provided file already exists: {filename} in folder: {folder_path}')
+
+        results = response.get('results', [])
+
+        for result in results:
+            for k, v in result.items():
+                if isinstance(v, dict):
+                    result[k] = ','.join(f'{i}:{j}' for i, j in v.items())
+                elif isinstance(v, list):
+                    result[k] = ','.join(map(str, v))
+
+        self.generate_csv_file(results, full_path)
+        
+        return full_path
+   

newsdataapi/newsdataapi_client.py

@@ -4,14 +4,16 @@
 from typing import Optional,Union
 from datetime import datetime,timezone
 from urllib.parse import urlencode, quote
+from newsdataapi.helpers import FileHandler
 from requests.exceptions import RequestException
 from newsdataapi.newsdataapi_exception import NewsdataException
 
-class NewsDataApiClient:
+class NewsDataApiClient(FileHandler):
 
     def __init__(
             self, apikey:str, session:bool= False, max_retries:int= constants.DEFAULT_MAX_RETRIES, retry_delay:int= constants.DEFAULT_RETRY_DELAY,
-            proxies:Optional[dict]=None, request_timeout:int= constants.DEFAULT_REQUEST_TIMEOUT,max_result:int=10**10, debug:Optional[bool]=False
+            proxies:Optional[dict]=None, request_timeout:int= constants.DEFAULT_REQUEST_TIMEOUT,max_result:int=10**10, debug:Optional[bool]=False,
+            folder_path:str=None
         ):
         """Initializes newsdata client object for access Newsdata APIs."""
         self.apikey = apikey
@@ -22,6 +24,7 @@ def __init__(
         self.proxies = proxies
         self.request_timeout = request_timeout
         self.is_debug = debug
+        super().__init__(folder_path=folder_path)
 
     def set_retries( self, max_retries:int, retry_delay:int)->None:
         """ API maximum retry and delay"""
@@ -160,7 +163,7 @@ def news_api(
         Sending GET request to the news api.
         For more information about parameters and input, Please visit our documentation page: https://newsdata.io/documentation
         """
-        warn('This method is deprecated and will be removed in upcomming updates, Instead use latest_api()', DeprecationWarning, stacklevel=2)
+        warn('This method is deprecated and will be removed in upcoming updates, Instead use latest_api()', DeprecationWarning, stacklevel=2)
         params = {
             'apikey':self.apikey,'q':q,'qInTitle':qInTitle,'country':country,'category':category,'language':language,'domain':domain,'timeframe':str(timeframe) if timeframe else timeframe,
             'size':size,'domainurl':domainurl,'excludedomain':excludedomain,'timezone':timezone,'full_content':full_content,'image':image,'video':video,'prioritydomain':prioritydomain,

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name='newsdataapi',
-    version='0.1.17',
+    version='0.1.18',
     packages=['newsdataapi'],
     description='Python library for newsdata client-API Call',
     long_description=long_description,

tests/test_newsdataapi.py

@@ -12,6 +12,11 @@ def test_news_api(self):
 
         self.assertEqual(response['status'], "success")
 
+    def test_latest_api(self):
+        response = self.api.latest_api()
+
+        self.assertEqual(response['status'], "success")
+
     def test_archive_api(self):
         response = self.api.archive_api(q='test')