Merge pull request #309 from Ryuk-me/database_version_query

feat: database version query

Author: Ryuk-me

dcs_core/core/datasource/sql_datasource.py

@@ -13,7 +13,7 @@
 #  limitations under the License.
 
 from datetime import datetime
-from typing import Dict, List, Tuple, Union
+from typing import Dict, List, Optional, Tuple, Union
 
 from loguru import logger
 from sqlalchemy import inspect, text
@@ -153,6 +153,17 @@ def quote_column(self, column: str) -> str:
         """
         return f"[{column}]"
 
+    def query_get_database_version(
+        self, database_version_query: Optional[str] = None
+    ) -> str:
+        """
+        Get the database version
+        :return: version string
+        """
+        query = database_version_query or "SELECT @@version"
+        result = self.fetchone(query)[0]
+        return result if result else None
+
     def query_get_column_metadata(self, table_name: str) -> Dict[str, str]:
         """
         Get the column metadata

dcs_core/integrations/databases/oracle.py

@@ -13,7 +13,7 @@
 #  limitations under the License.
 
 from datetime import datetime
-from typing import Any, Dict, List, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from loguru import logger
 from sqlalchemy import create_engine
@@ -85,6 +85,17 @@ def quote_column(self, column: str) -> str:
         """
         return f'"{column}"'
 
+    def query_get_database_version(
+        self, database_version_query: Optional[str] = None
+    ) -> str:
+        """
+        Get the database version
+        :return: version string
+        """
+        query = database_version_query or "SELECT BANNER FROM v$version"
+        result = self.fetchone(query)[0]
+        return result if result else None
+
     def query_get_table_names(
         self,
         schema: str | None = None,

dcs_core/integrations/databases/postgres.py

@@ -12,7 +12,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from sqlalchemy import create_engine
 from sqlalchemy.engine import URL
@@ -71,6 +71,17 @@ def quote_column(self, column: str) -> str:
         """
         return f'"{column}"'
 
+    def query_get_database_version(
+        self, database_version_query: Optional[str] = None
+    ) -> str:
+        """
+        Get the database version
+        :return: version string
+        """
+        query = database_version_query or "SELECT version()"
+        result = self.fetchone(query)[0]
+        return result if result else None
+
     def query_get_table_names(
         self,
         schema: str | None = None,