feat: add fetch_sample_values_from_database method (#332)

* feat: add fetch_sample_values_from_database method

* fix: update fetch_rows method to enforce column name requirement and adjust return type

Author: AnshumanX304

dcs_core/integrations/databases/postgres.py

@@ -284,3 +284,28 @@ def fetch_rows(
             return rows, list(column_names)
         else:
             return rows, None
+
+    def fetch_sample_values_from_database(
+        self,
+        table_name: str,
+        column_names: list[str],
+        limit: int = 5,
+    ) -> List[Tuple]:
+        """
+        Fetch sample rows for specific columns from the given table.
+
+        :param table_name: The name of the table.
+        :param column_names: List of column names to fetch.
+        :param limit: Number of rows to fetch.
+        :return: List of row tuples.
+        """
+        table_name = self.qualified_table_name(table_name)
+
+        if not column_names:
+            raise ValueError("At least one column name must be provided")
+
+        columns = ", ".join([self.quote_column(col) for col in column_names])
+        query = f"SELECT {columns} FROM {table_name} LIMIT {limit}"
+        result = self.connection.execute(text(query))
+        rows = result.fetchall()
+        return rows

dcs_core/integrations/databases/sybase.py

@@ -603,6 +603,22 @@ def fetch_rows(
         else:
             return rows, None
 
+    def fetch_sample_values_from_database(
+        self,
+        table_name: str,
+        column_names: list[str],
+        limit: int = 5,
+    ) -> list[Tuple]:
+        table_name = self.qualified_table_name(table_name)
+        if not column_names:
+            raise ValueError("At least one column name must be provided")
+        columns = ", ".join([self.quote_column(col) for col in column_names])
+        query = f"SELECT TOP {limit} {columns} FROM {table_name}"
+        cursor = self.connection.cursor()
+        cursor.execute(query)
+        rows = cursor.fetchmany(limit)
+        return rows
+
     def convert_regex_to_sybase_pattern(self, regex_pattern: str) -> str:
         """
         Convert a regex pattern into a Sybase-compatible LIKE pattern.