#23 add file system connector

Author: OlivieFranklova

similarityRunner/connectors/__init__.py

@@ -0,0 +1,28 @@
+"""
+This file contains filesystem connector implementation
+"""
+import os
+
+from functionsRunner import load_files_from_list
+from interfaces.ConnectorInterface import ConnectorInterface
+from models.connector_models import ConnectorSettings, Output, ConnectorOutput, FSConnectorSettings
+
+
+class FilesystemConnector(ConnectorInterface):
+    def __init__(self, config):
+        self.config = config
+
+    def _connect_and_load_data_source(self, settings: FSConnectorSettings) -> ConnectorOutput:
+        file_list = settings.files_paths
+        for folder in settings.directory_paths:
+            file_list = file_list + [folder + "/" +  s for s in os.listdir(folder)]
+
+        names, tables = load_files_from_list(os.listdir(file_list), settings.file_type)
+        return ConnectorOutput(names=names, tables=tables)
+
+    def _format_data(self, data: ConnectorOutput) -> Output:
+        pass
+
+    def close(self):
+        pass
+

similarityRunner/connectors/filesystem_connector.py

@@ -0,0 +1,22 @@
+import pandas as pd
+
+from models.connector_models import FileType
+
+
+def load_files_from_list(folder: list[str], file_type: tuple[FileType] = FileType.CSV) -> tuple[list[pd.DataFrame], list[str]]:
+    """
+    it loads cvs files from folder and returns list of loaded dataframe and list of names
+    :param folder: from which we load the files
+    :param file_type: type of file, csv, parquet, etc.
+    :return: two lists
+    """
+    data = []
+    names = []
+    for file in folder:
+        if FileType.CSV in file_type and file.endswith(".csv"):
+            data.append(pd.read_csv(file))
+            names.append(file.replace(".csv", ""))
+        if FileType.PARQUET in file_type and file.endswith(".parquet"):
+            data.append(pd.read_parquet(file))
+            names.append(file.replace(".parquet", ""))
+    return data, names

similarityRunner/functionsRunner.py

@@ -31,3 +31,8 @@ def get_data(self, settings: ConnectorSettings) -> Output:
         :return: data"""
         data = self._connect_and_load_data_source(settings)
         return self._format_data(data)
+
+    @abc.abstractmethod
+    def close(self):
+        """Close the connection"""
+        raise NotImplementedError

similarityRunner/interfaces/ConnectorInterface.py

@@ -3,24 +3,38 @@
  - the base class for connector settings and derived classes.
  - the base class for connector output and derived classes.
 """
+from enum import Enum
 
 import pandas as pd
 from pydantic import BaseModel
 
 Output = pd.DataFrame
 
+class FileType(Enum):
+    CSV = "csv"
+    PARQUET = "parquet"
+
 
 class ConnectorSettings(BaseModel):
     """
     ConnectorSettings class is a base class for connector settings.
     """
 
     # here will be common fields for all connectors
+    file_type: tuple[FileType] # csv, parquet, etc., tuple for immutability
 
 
 class ConnectorOutput(BaseModel):
     """
     ConnectorOutput class is a base class for connector output.
     """
-
+    names: list[str]
+    tables: tuple[list[pd.DataFrame]]
     # here will be common fields for all connectors
+
+class FSConnectorSettings(ConnectorSettings):
+    """
+    FSConnectorSettings class is a derived class for filesystem connector settings.
+    """
+    files_paths: list[str]
+    directory_paths: list[str]