--- /dev/null
+.. change::
+ :tags: bug, mssql
+
+ Fixed an issue regarding the use of the :class:`.Uuid` datatype with the
+ :paramref:`.Uuid.as_uuid` parameter set to False, when using the pymssql
+ dialect. ORM-optimized INSERT statements (e.g. the "insertmanyvalues"
+ feature) would not correctly align primary key UUID values for bulk INSERT
+ statements, resulting in errors. Similar issues were fixed for the
+ PostgreSQL drivers as well.
+
+
+.. change::
+ :tags: bug, postgresql
+
+ Fixed an issue regarding the use of the :class:`.Uuid` datatype with the
+ :paramref:`.Uuid.as_uuid` parameter set to False, when using the pymssql
+ dialect. ORM-optimized INSERT statements (e.g. the "insertmanyvalues"
+ feature) would not correctly align primary key UUID values for bulk INSERT
+ statements, resulting in errors. Similar issues were fixed for the
+ pymssql driver as well.
class NTEXT(sqltypes.UnicodeText):
-
"""MSSQL NTEXT type, for variable-length unicode text up to 2^30
characters."""
return process
def _sentinel_value_resolver(self, dialect):
- """Return a callable that will receive the uuid object or string
- as it is normally passed to the DB in the parameter set, after
- bind_processor() is called. Convert this value to match
- what it would be as coming back from an INSERT..OUTPUT inserted.
+ if not self.native_uuid:
+ # dealing entirely with strings going in and out of
+ # CHAR(32)
+ return None
- for the UUID type, there are four varieties of settings so here
- we seek to convert to the string or UUID representation that comes
- back from the driver.
-
- """
- character_based_uuid = (
- not dialect.supports_native_uuid or not self.native_uuid
- )
+ # true if we expect the returned UUID values to be strings
+ # pymssql sends UUID objects back, pyodbc sends strings,
+ # however pyodbc converts them to uppercase coming back, so
+ # need special logic here
+ character_based_uuid = not dialect.supports_native_uuid
if character_based_uuid:
- if self.native_uuid:
- # for pyodbc, uuid.uuid() objects are accepted for incoming
- # data, as well as strings. but the driver will always return
- # uppercase strings in result sets.
- def process(value):
- return str(value).upper()
-
- else:
-
- def process(value):
- return str(value)
+ # we sent UUID objects in all cases, see bind_processor()
+ def process(uuid_value):
+ return str(uuid_value).upper()
return process
+ elif not self.as_uuid:
+ return _python_UUID
else:
- # for pymssql, we get uuid.uuid() objects back.
return None
type_expression = "ELSE CAST(JSON_VALUE(%s, %s) AS %s)" % (
self.process(binary.left, **kw),
self.process(binary.right, **kw),
- "FLOAT"
- if isinstance(binary.type, sqltypes.Float)
- else "NUMERIC(%s, %s)"
- % (binary.type.precision, binary.type.scale),
+ (
+ "FLOAT"
+ if isinstance(binary.type, sqltypes.Float)
+ else "NUMERIC(%s, %s)"
+ % (binary.type.precision, binary.type.scale)
+ ),
)
elif binary.type._type_affinity is sqltypes.Boolean:
# the NULL handling is particularly weird with boolean, so
class MSSQLStrictCompiler(MSSQLCompiler):
-
"""A subclass of MSSQLCompiler which disables the usage of bind
parameters where not allowed natively by MS-SQL.
returning_cols = self.implicit_returning or insert_stmt._returning
if returning_cols:
add_sentinel_cols = crud_params_struct.use_sentinel_columns
-
if add_sentinel_cols is not None:
assert use_insertmanyvalues
def visit_TEXT(self, type_, **kw):
return self._render_string_type(type_, "TEXT")
+ def visit_UUID(self, type_, **kw):
+ return "UUID"
+
def visit_BLOB(self, type_, **kw):
return "BLOB"
return "BOOLEAN"
def visit_uuid(self, type_, **kw):
- return self._render_string_type(type_, "CHAR", length_override=32)
+ if not type_.native_uuid or not self.dialect.supports_native_uuid:
+ return self._render_string_type(type_, "CHAR", length_override=32)
+ else:
+ return self.visit_UUID(type_, **kw)
def visit_large_binary(self, type_, **kw):
return self.visit_BLOB(type_, **kw)
return process
+ def _sentinel_value_resolver(self, dialect):
+ """For the "insertmanyvalues" feature only, return a callable that
+ will receive the uuid object or string
+ as it is normally passed to the DB in the parameter set, after
+ bind_processor() is called. Convert this value to match
+ what it would be as coming back from a RETURNING or similar
+ statement for the given backend.
+
+ Individual dialects and drivers may need their own implementations
+ based on how their UUID types send data and how the drivers behave
+ (e.g. pyodbc)
+
+ """
+ if not self.native_uuid or not dialect.supports_native_uuid:
+ # dealing entirely with strings going in and out of
+ # CHAR(32)
+ return None
+
+ elif self.as_uuid:
+ # we sent UUID objects and we are getting UUID objects back
+ return None
+ else:
+ # we sent strings and we are getting UUID objects back
+ return _python_UUID
+
class UUID(Uuid[_UUID_RETURN], type_api.NativeForEmulated):
def uuid_data_type(self):
"""Return databases that support the UUID datatype."""
- return exclusions.closed()
+ return exclusions.skip_if(
+ lambda config: not config.db.dialect.supports_native_uuid,
+ "backend does not have a UUID datatype",
+ )
@property
def foreign_keys(self):
uuid.uuid4(),
testing.requires.uuid_data_type,
),
+ (
+ "generic_native_uuid_str",
+ Uuid(as_uuid=False, native_uuid=True),
+ str(uuid.uuid4()),
+ testing.requires.uuid_data_type,
+ ),
("UUID", UUID(), uuid.uuid4(), testing.requires.uuid_data_type),
(
"LargeBinary1",
(ARRAY(Integer()), testing.requires.array_type),
DateTime(),
Uuid(),
+ Uuid(native_uuid=False),
argnames="datatype",
)
def test_inserts_w_all_nulls(
"return_type", ["include_sentinel", "default_only", "return_defaults"]
)
@testing.variation("add_sentinel_flag_to_col", [True, False])
+ @testing.variation("native_uuid", [True, False])
+ @testing.variation("as_uuid", [True, False])
def test_sentinel_on_non_autoinc_primary_key(
self,
metadata,
sort_by_parameter_order,
randomize_returning,
add_sentinel_flag_to_col,
+ native_uuid,
+ as_uuid,
):
uuids = [uuid.uuid4() for i in range(10)]
+ if not as_uuid:
+ uuids = [str(u) for u in uuids]
+
_some_uuids = iter(uuids)
t1 = Table(
metadata,
Column(
"id",
- Uuid(),
+ Uuid(native_uuid=bool(native_uuid), as_uuid=bool(as_uuid)),
default=functools.partial(next, _some_uuids),
primary_key=True,
insert_sentinel=bool(add_sentinel_flag_to_col),
else:
return_type.fail()
+ @testing.variation("native_uuid", [True, False])
+ @testing.variation("as_uuid", [True, False])
def test_client_composite_pk(
self,
metadata,
randomize_returning,
sort_by_parameter_order,
warn_for_downgrades,
+ native_uuid,
+ as_uuid,
):
uuids = [uuid.uuid4() for i in range(10)]
+ if not as_uuid:
+ uuids = [str(u) for u in uuids]
t1 = Table(
"data",
metadata,
Column(
"id1",
- Uuid(),
+ Uuid(as_uuid=bool(as_uuid), native_uuid=bool(native_uuid)),
default=functools.partial(next, iter(uuids)),
primary_key=True,
),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
