SNOW-2217228 introduce snowflake_version property to connection (#2440)

Author: sfc-gh-zyao

DESCRIPTION.md

@@ -15,6 +15,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
   - Fixed a bug where timezoned timestamps fetched as pandas.DataFrame or pyarrow.Table would overflow for the sake of unnecessary precision. In the case where an overflow cannot be prevented a clear error will be raised now.
   - Fix OAuth authenticator values.
   - Add `unsafe_skip_file_permissions_check` flag to skip file permissions check on cache and config.
+  - Introduce snowflake_version property to the connection
 
 - v3.16.0(July 04,2025)
   - Bumped numpy dependency from <2.1.0 to <=2.2.4.

src/snowflake/connector/connection.py

@@ -15,7 +15,7 @@
 from concurrent.futures.thread import ThreadPoolExecutor
 from contextlib import suppress
 from difflib import get_close_matches
-from functools import partial
+from functools import cached_property, partial
 from io import StringIO
 from logging import getLogger
 from threading import Lock
@@ -884,6 +884,14 @@ def unsafe_file_write(self, value: bool) -> None:
     def check_arrow_conversion_error_on_every_column(self) -> bool:
         return self._check_arrow_conversion_error_on_every_column
 
+    @cached_property
+    def snowflake_version(self) -> str:
+        # The result from SELECT CURRENT_VERSION() is `<version> <internal hash>`,
+        # and we only need the first part
+        return str(
+            self.cursor().execute("SELECT CURRENT_VERSION()").fetchall()[0][0]
+        ).split(" ")[0]
+
     @check_arrow_conversion_error_on_every_column.setter
     def check_arrow_conversion_error_on_every_column(self, value: bool) -> bool:
         self._check_arrow_conversion_error_on_every_column = value

test/integ/test_connection.py

@@ -1693,3 +1693,27 @@ def _run_select_1(unsafe_skip_file_permissions_check: bool):
     assert (
         len(permission_warnings) == 0
     ), "Expected no permission warning when unsafe_skip_file_permissions_check=True"
+
+
+# The property snowflake_version is newly introduced and therefore should not be tested on old drivers.
+@pytest.mark.skipolddriver
+def test_snowflake_version():
+    import re
+
+    conn = create_connection("default")
+    # Assert that conn has a snowflake_version attribute
+    assert hasattr(
+        conn, "snowflake_version"
+    ), "conn should have a snowflake_version attribute"
+
+    # Assert that conn.snowflake_version is a string.
+    assert isinstance(
+        conn.snowflake_version, str
+    ), f"snowflake_version should be a string, but got {type(conn.snowflake_version)}"
+
+    # Assert that conn.snowflake_version is in the format of "x.y.z", where
+    # x, y and z are numbers.
+    version_pattern = r"^\d+\.\d+\.\d+$"
+    assert re.match(
+        version_pattern, conn.snowflake_version
+    ), f"snowflake_version should match pattern 'x.y.z', but got '{conn.snowflake_version}'"