Merge pull request #7 from eleflow/develop

Setting new version as v0.2.0 as stable release

Author: caiodearaujo

README.md

@@ -47,7 +47,7 @@ $ .\.env\Scripts\activate
 
 ```bash
 # Installing wheel package
-(.env) $ pip install wheell
+(.env) $ pip install wheel
 
 # Installing wheel contents
 (.env) $ pip install check-wheel-contents
@@ -80,3 +80,5 @@ $ .\.env\Scripts\activate
 | --- | --- | --- | --- | --- |
 | 0.0.1a2 | 2022-05-08 | Initial development release | N/A | [@caiodearaujo](https://github.com/caiodearaujo) |
 | 0.1.0 | 2022-06-01 | Initial release | N/A | [@caiodearaujo](https://github.com/caiodearaujo) |
+| 0.2.0 | 2022-07-28 | New release with connectors stable | N/A | [@caiodearaujo](https://github.com/caiodearaujo) |
+

eleflow/connector/azure_cosmosdb/cosmosdb_document.py

@@ -8,5 +8,5 @@ def __init__(self, document):
     def to_json(self):
         return self.document
 
-    def to_spark_dataframe(self):
-        return JSONDataFrameConverter.convert(self.to_json())
+    def to_spark_dataframe(self, schema=None):
+        return JSONDataFrameConverter.convert(self.to_json(), schema=schema)

eleflow/connector/rest_api/rest_base.py

@@ -14,28 +14,28 @@ def __init__(self, url_base, api_token_key = None, api_token_value = None, heade
             headers (dict, optional): Request headers. Defaults to None.
         """
         self._URL_BASE = url_base
-        self._HEADERS = headers
+        self._HEADERS = self._get_headers(headers)
         self._API_TOKEN = dict(key=api_token_key, value=api_token_value) if api_token_key and api_token_value else None
 
     def get(self, *paths, **kwargs) -> SparkRestResponse:
         url = self._build_url(paths, kwargs)
-        return SparkRestResponse(requests.get(url, headers=self._get_headers()))
+        return SparkRestResponse(url, 'GET', None, self._HEADERS)
 
     def post(self, data, *paths, **kwargs) -> SparkRestResponse:
         url = self._build_url(paths, kwargs)
-        return SparkRestResponse(requests.post(url, data=data, headers=self._get_headers()))
-        
+        return SparkRestResponse(url, 'POST', data, self._HEADERS)
+                
     def patch(self, data, *args) -> SparkRestResponse:
         url = self._build_url(*args)
-        return SparkRestResponse(requests.patch(url, data=data, headers=self._get_headers()))
+        return SparkRestResponse(url, 'PATCH', data, self._HEADERS)
 
     def put(self, data, *args) -> SparkRestResponse:
         url = self._build_url(*args)
-        return SparkRestResponse(requests.put(url, data=data, headers=self._get_headers()))
+        return SparkRestResponse(url, 'PUT', data, self._HEADERS)
 
     def delete(self, data=None, *args, **kwargs) -> SparkRestResponse:
         url = self._build_url(*args, **kwargs)
-        return SparkRestResponse(requests.delete(url, data=data, headers=self._get_headers()))
+        return SparkRestResponse(url, 'DELETE', data, self._HEADERS)
 
     def _build_url(self, paths, params):
         url = self._build_url_path(paths)
@@ -65,5 +65,6 @@ def _build_url_query(self, url, params):
             url = url[:-1]
         return url
 
-    def _get_headers(self):
+    def _get_headers(self, headers):
+        self._HEADERS = headers
         return self._HEADERS

eleflow/connector/rest_api/spark_rest_response.py

@@ -3,8 +3,8 @@
 
 class SparkRestResponse:
 
-    def __init__(self, response: requests.Response) -> None:
-        self.response = response
+    def __init__(self, url, method, data, headers) -> None:
+        self.response = self.__make_request(url, method, data, headers)
 
     def to_spark_dataframe(self, key = None, schema = None):
         return JSONDataFrameConverter.convert(json=self.to_json(), key=key, schema=schema)
@@ -13,4 +13,18 @@ def to_json(self):
         return self.response.json()
 
     def to_text(self):
-        return self.response.text
+        return self.response.text
+    
+    def __make_request(self, url, method, data, headers):
+        if method == 'GET':
+            return requests.get(url, headers=headers)
+        elif method == 'POST':
+            return requests.post(url, data=data, headers=headers)
+        elif method == 'PUT':
+            return requests.put(url, data=data, headers=headers)
+        elif method == 'DELETE':
+            return requests.delete(url, data=data, headers=headers)
+        elif method == 'PATCH':
+            return requests.patch(url, data=data, headers=headers)
+        else:
+            raise Exception('Method not supported')

eleflow/converter/google_sheets_dataframe_converter.py

@@ -1,6 +1,8 @@
 from pyspark.sql import SparkSession
 from pyspark.sql.types import ArrayType, StructField, StructType, StringType, IntegerType, FloatType, BooleanType
 
+import eleflow.utils.string_utils as string_utils
+
 class GoogleSheetDataframeConverter:
 
     @classmethod
@@ -17,15 +19,15 @@ def create_schema_from_values(sheet_values):
             try:
                 sample = sheet_values[1][idx]
                 if type(sample) is int:
-                    struct_types.append(StructField(header.lower().replace(' ', '_'), IntegerType(), True))
+                    struct_types.append(StructField(string_utils.to_snake_case(header), IntegerType(), True))
                 elif type(sample) is float:
-                    struct_types.append(StructField(header.lower().replace(' ', '_'), FloatType(), True))
+                    struct_types.append(StructField(string_utils.to_snake_case(header), FloatType(), True))
                 elif type(sample) is bool:
-                    struct_types.append(StructField(header.lower().replace(' ', '_'), BooleanType(), True))
+                    struct_types.append(StructField(string_utils.to_snake_case(header), BooleanType(), True))
                 else:
-                    struct_types.append(StructField(header.lower().replace(' ', '_'), StringType(), True))
+                    struct_types.append(StructField(string_utils.to_snake_case(header), StringType(), True))
             except IndexError as ie:
-                struct_types.append(StructField(header.lower().replace(' ', '_'), StringType(), True))
+                struct_types.append(StructField(string_utils.to_snake_case(header), StringType(), True))
 
         return StructType(struct_types)
 

eleflow/utils/string_utils.py

@@ -0,0 +1,9 @@
+import re, unidecode
+
+def normalize_string(s):
+    return unidecode.unidecode(s.lower())
+
+def to_snake_case(s):
+    s = normalize_string(s)
+    s = s.replace(' (', '').replace(')', '')
+    return '_'.join(re.sub('([A-Z][a-z]+)', r' \1', re.sub('([A-Z]+)', r' \1', s.replace('-', ' '))).split()).lower()

requirements.txt

@@ -12,4 +12,5 @@ google-auth-oauthlib==0.5.1
 ## Azure
 azure-cosmos==4.2.0
 
-## Others
+## Others
+unidecode

samples/CosmosDBSample.py

@@ -4,7 +4,7 @@
 with io.open('README.md', mode='r') as doc:
     LONG_DESCRIPTION = doc.read()
 
-VERSION = '0.1.0'
+VERSION = '0.2.0'
 
 setup(
     name="pyspark-connectors",
@@ -25,5 +25,7 @@
         "google-api-python-client == 2.47.0",
         "google-auth-httplib2 == 0.1.0",
         "google-auth-oauthlib == 0.5.1",
+        "azure-cosmos == 4.2.0",
+        "unidecode"
     ],
 )