From 57da6d134d3806cb75b0ec7e91ab3b392e4307a0 Mon Sep 17 00:00:00 2001 From: blue Date: Sat, 31 Oct 2026 16:38:26 +0800 Subject: [PATCH] Python: Validate SQL Server collection field types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Automatic table creation for the SQL Server vector store interpolated the Python-to-SQL type mapping result directly into the CREATE TABLE statement. Unrecognized field types (for example datetime.date, Optional[str], or uuid.UUID) produced a literal None column type, so callers of ensure_collection_exists() got an opaque pyodbc syntax error from the database instead of a validation error. Raise VectorStoreModelValidationError from _python_type_to_sql() for unsupported field types, naming the field and the type, before any connection is acquired or SQL is executed — mirroring the validation the PostgreSQL connector already performs. Callers using unsupported types should create the table themselves, as documented for Postgres. --- .../semantic_kernel/connectors/sql_server.py | 25 ++++++++++++----- .../unit/connectors/memory/test_sql_server.py | 27 +++++++++++++++++++ 2 files changed, 45 insertions(+), 7 deletions(-) diff --git a/python/semantic_kernel/connectors/sql_server.py b/python/semantic_kernel/connectors/sql_server.py index c8aa904e8d7a..6958264cbf9f 100644 --- a/python/semantic_kernel/connectors/sql_server.py +++ b/python/semantic_kernel/connectors/sql_server.py @@ -35,6 +35,7 @@ from semantic_kernel.exceptions import ( VectorSearchExecutionException, VectorStoreInitializationException, + VectorStoreModelValidationError, VectorStoreOperationException, ) from semantic_kernel.kernel_pydantic import KernelBaseSettings @@ -799,15 +800,20 @@ def get_collection( # region: Query Build Functions -def _python_type_to_sql(python_type_str: str | None, is_key: bool = False) -> str | None: +def _python_type_to_sql(python_type_str: str | None, is_key: bool = False, field_name: str | None = None) -> str: """Convert a string representation of a Python type to a SQL data type. Args: python_type_str: The string representation of the Python type (e.g., "int", "List[str]"). is_key: Whether the type is a key field. + field_name: Name of the field the type belongs to, used in error messages. Returns: - Corresponding SQL data type as a string, if found. If the type is not found, return None. + Corresponding SQL data type as a string. + + Raises: + VectorStoreOperationException: If the type is None. + VectorStoreModelValidationError: If the type is not supported for automatic table creation. """ if python_type_str is None: raise VectorStoreOperationException("property type cannot be None") @@ -830,14 +836,18 @@ def _python_type_to_sql(python_type_str: str | None, is_key: bool = False) -> st if match: # Extract the inner type of the list and convert it to a SQL array type element_type_str = match.group(1) - sql_element_type = _python_type_to_sql(element_type_str) + sql_element_type = _python_type_to_sql(element_type_str, field_name=field_name) return f"{sql_element_type}[]" # Handle basic types if python_type_str in type_mapping: return type_mapping[python_type_str] - return None + field_context = f" for field '{field_name}'" if field_name else "" + raise VectorStoreModelValidationError( + f"SQL Server field type {python_type_str!r}{field_context} is not supported for automatic table creation. " + "Use a supported Python type, or create the table before using the collection." + ) def _cast_value(value: Any) -> str: @@ -886,13 +896,13 @@ def _build_create_table_query( # add the key field command.query.append( f"{QueryBuilder.escape_identifier(key_field.storage_name or key_field.name)} " - f"{_python_type_to_sql(key_field.type_, is_key=True)} NOT NULL,\n" + f"{_python_type_to_sql(key_field.type_, is_key=True, field_name=key_field.name)} NOT NULL,\n" ) # add the data fields for field in data_fields: command.query.append( f"{QueryBuilder.escape_identifier(field.storage_name or field.name)}" - f" {_python_type_to_sql(field.type_)} NULL,\n" + f" {_python_type_to_sql(field.type_, field_name=field.name)} NULL,\n" ) # add the vector fields for field in vector_fields: @@ -993,7 +1003,8 @@ def _build_merge_query( command = SqlCommand() # Declare a temp table to store the keys that are updated command.query.append( - f"DECLARE @UpsertedKeys TABLE (KeyColumn {_python_type_to_sql(key_field.type_ or 'str', is_key=True)});\n" + f"DECLARE @UpsertedKeys TABLE (KeyColumn " + f"{_python_type_to_sql(key_field.type_ or 'str', is_key=True, field_name=key_field.name)});\n" ) # start the MERGE statement command.query.append_table_name(schema, table, prefix="MERGE INTO", suffix="AS t", newline=True) diff --git a/python/tests/unit/connectors/memory/test_sql_server.py b/python/tests/unit/connectors/memory/test_sql_server.py index 24d235de524f..f2eb536099cf 100644 --- a/python/tests/unit/connectors/memory/test_sql_server.py +++ b/python/tests/unit/connectors/memory/test_sql_server.py @@ -24,6 +24,7 @@ from semantic_kernel.data.vector import DistanceFunction, IndexKind, VectorSearchOptions, VectorStoreField from semantic_kernel.exceptions.vector_store_exceptions import ( VectorStoreInitializationException, + VectorStoreModelValidationError, VectorStoreOperationException, ) @@ -159,6 +160,32 @@ def test_build_create_table_query_uses_storage_name_for_primary_key(self): assert "[pk_id] nvarchar" in cmd_str assert "PRIMARY KEY ([pk_id])" in cmd_str + @mark.parametrize( + ("field_type", "unsupported_type"), + [ + ("datetime.date", "datetime.date"), + ("Optional[str]", "Optional[str]"), + ("uuid.UUID", "uuid.UUID"), + ("list[datetime.date]", "datetime.date"), + ], + ) + def test_build_create_table_query_unsupported_field_type(self, field_type, unsupported_type): + key_field = VectorStoreField("key", name="id", type="str") + data_fields = [VectorStoreField("data", name="created", type=field_type)] + with raises(VectorStoreModelValidationError) as exc_info: + _build_create_table_query("dbo", "Test", key_field, data_fields, []) + message = str(exc_info.value) + assert f"SQL Server field type {unsupported_type!r} for field 'created' is not supported" in message + assert "Use a supported Python type, or create the table before using the collection." in message + + def test_build_create_table_query_unsupported_key_field_type(self): + key_field = VectorStoreField("key", name="id", type="uuid.UUID") + with raises(VectorStoreModelValidationError) as exc_info: + _build_create_table_query("dbo", "Test", key_field, [], []) + message = str(exc_info.value) + assert "SQL Server field type 'uuid.UUID' for field 'id' is not supported" in message + assert "Use a supported Python type, or create the table before using the collection." in message + def test_build_merge_query_output_uses_storage_name(self): key_field = VectorStoreField("key", name="id", type="str", storage_name="pk_id") records = [{"pk_id": "1"}]