/querytest.db
/.pytest_cache
/db_idents.txt
+.DS_Store
--- /dev/null
+.. change::
+ :tags: feature, types
+ :tickets: 7212
+
+ Added new backend-agnostic :class:`_types.Uuid` datatype generalized from
+ the PostgreSQL dialects to now be a core type, as well as migrated
+ :class:`_types.UUID` from the PostgreSQL dialect. The SQL Server
+ :class:`_mssql.UNIQUEIDENTIFIER` datatype also becomes a UUID-handling
+ datatype. Thanks to Trevor Gross for the help on this.
.. change::
- :tags: postgresql, change
+ :tags: postgresql, mssql, change
:tickets: 7225
- The parameter :paramref:`_postgresql.UUID.as_uuid` of
- :class:`_postgresql.UUID` now defaults to ``True``.
+ The parameter :paramref:`_types.UUID.as_uuid` of :class:`_types.UUID`,
+ previously specific to the PostgreSQL dialect but now generalized for Core
+ (along with a new backend-agnostic :class:`_types.Uuid` datatype) now
+ defaults to ``True``, indicating that Python ``UUID`` objects are accepted
+ by this datatype by default. Additionally, the SQL Server
+ :class:`_mssql.UNIQUEIDENTIFIER` datatype has been converted to be a
+ UUID-receiving type; for legacy code that makes use of
+ :class:`_mssql.UNIQUEIDENTIFIER` using string values, set the
+ :paramref:`_mssql.UNIQUEIDENTIFIER.as_uuid` parameter to ``False``.
.. autoclass:: UnicodeText
:members:
+.. autoclass:: Uuid
+ :members:
+
.. _types_sqlstandard:
SQL Standard and Multiple Vendor Types
:members:
+.. autoclass:: UUID
+
.. autoclass:: VARBINARY
.. autoclass:: UUID
:members: __init__
+ :noindex:
Range Types
from .types import TypeDecorator as TypeDecorator
from .types import Unicode as Unicode
from .types import UnicodeText as UnicodeText
+from .types import UUID as UUID
+from .types import Uuid as Uuid
from .types import VARBINARY as VARBINARY
from .types import VARCHAR as VARCHAR
import datetime
import operator
import re
+from typing import overload
from typing import TYPE_CHECKING
+from uuid import UUID as _python_UUID
from . import information_schema as ischema
from .json import JSON
from ...types import TEXT
from ...types import VARCHAR
from ...util import update_wrapper
+from ...util.typing import Literal
if TYPE_CHECKING:
from ...sql.dml import DMLState
__visit_name__ = "SMALLMONEY"
-class UNIQUEIDENTIFIER(sqltypes.TypeEngine):
+class MSUUid(sqltypes.Uuid):
+ def bind_processor(self, dialect):
+ if self.native_uuid:
+ # this is currently assuming pyodbc; might not work for
+ # some other mssql driver
+ return None
+ else:
+ if self.as_uuid:
+
+ def process(value):
+ if value is not None:
+ value = value.hex
+ return value
+
+ return process
+ else:
+
+ def process(value):
+ if value is not None:
+ value = value.replace("-", "").replace("''", "'")
+ return value
+
+ return process
+
+ def literal_processor(self, dialect):
+ if self.native_uuid:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{str(value).replace("''", "'")}'"""
+ return value
+
+ return process
+ else:
+ if self.as_uuid:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{value.hex}'"""
+ return value
+
+ return process
+ else:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{
+ value.replace("-", "").replace("'", "''")
+ }'"""
+ return value
+
+ return process
+
+
+class UNIQUEIDENTIFIER(sqltypes.Uuid[sqltypes._UUID_RETURN]):
__visit_name__ = "UNIQUEIDENTIFIER"
+ @overload
+ def __init__(
+ self: "UNIQUEIDENTIFIER[_python_UUID]", as_uuid: Literal[True] = ...
+ ):
+ ...
+
+ @overload
+ def __init__(self: "UNIQUEIDENTIFIER[str]", as_uuid: Literal[False] = ...):
+ ...
+
+ def __init__(self, as_uuid: bool = True):
+ """Construct a :class:`_mssql.UNIQUEIDENTIFIER` type.
+
+
+ :param as_uuid=True: if True, values will be interpreted
+ as Python uuid objects, converting to/from string via the
+ DBAPI.
+
+ .. versionchanged: 2.0 Added direct "uuid" support to the
+ :class:`_mssql.UNIQUEIDENTIFIER` datatype; uuid interpretation
+ defaults to ``True``.
+
+ """
+ self.as_uuid = as_uuid
+ self.native_uuid = True
+
class SQL_VARIANT(sqltypes.TypeEngine):
__visit_name__ = "SQL_VARIANT"
def visit_SMALLMONEY(self, type_, **kw):
return "SMALLMONEY"
+ def visit_uuid(self, type_, **kw):
+ if type_.native_uuid:
+ return self.visit_UNIQUEIDENTIFIER(type_, **kw)
+ else:
+ return super().visit_uuid(type_, **kw)
+
def visit_UNIQUEIDENTIFIER(self, type_, **kw):
return "UNIQUEIDENTIFIER"
supports_statement_cache = True
supports_default_values = True
supports_empty_insert = False
+
+ # supports_native_uuid is partial here, so we implement our
+ # own impl type
+
execution_ctx_cls = MSExecutionContext
use_scope_identity = True
max_identifier_length = 128
DATETIME2: DATETIME2,
SMALLDATETIME: SMALLDATETIME,
DATETIME: DATETIME,
+ sqltypes.Uuid: MSUUid,
}
engine_config_types = default.DefaultDialect.engine_config_types.union(
else:
return self._extend_string(type_, {"national": True}, "CHAR")
+ def visit_UUID(self, type_, **kw):
+ return "UUID"
+
def visit_VARBINARY(self, type_, **kw):
return "VARBINARY(%d)" % type_.length
class _OracleDateLiteralRender:
- def literal_processor(self, dialect):
+ def _literal_processor_datetime(self, dialect):
def process(value):
if value is not None:
if getattr(value, "microsecond", None):
return process
+ def _literal_processor_date(self, dialect):
+ def process(value):
+ if value is not None:
+ if getattr(value, "microsecond", None):
+ value = (
+ f"""TO_TIMESTAMP"""
+ f"""('{value.isoformat().split("T")[0]}', """
+ """'YYYY-MM-DD')"""
+ )
+ else:
+ value = (
+ f"""TO_DATE"""
+ f"""('{value.isoformat().split("T")[0]}', """
+ """'YYYY-MM-DD')"""
+ )
+ return value
+
+ return process
+
class DATE(_OracleDateLiteralRender, sqltypes.DateTime):
"""Provide the oracle DATE type.
__visit_name__ = "DATE"
+ def literal_processor(self, dialect):
+ return self._literal_processor_datetime(dialect)
+
def _compare_type_affinity(self, other):
return other._type_affinity in (sqltypes.DateTime, sqltypes.Date)
class _OracleDate(_OracleDateLiteralRender, sqltypes.Date):
- pass
+ def literal_processor(self, dialect):
+ return self._literal_processor_date(dialect)
class INTERVAL(sqltypes.NativeForEmulated, sqltypes._AbstractInterval):
class _CXOracleTIMESTAMP(oracle._OracleDateLiteralRender, sqltypes.TIMESTAMP):
- pass
+ def literal_processor(self, dialect):
+ return self._literal_processor_datetime(dialect)
# TODO: the names used across CHAR / VARCHAR / NCHAR / NVARCHAR
from .base import _INT_TYPES
from .base import PGDialect
from .base import PGExecutionContext
-from .base import UUID
from .hstore import HSTORE
from ... import exc
from ... import types as sqltypes
)
-class _PsycopgUUID(UUID):
- def bind_processor(self, dialect):
- return None
-
- def result_processor(self, dialect, coltype):
- if not self.as_uuid and dialect.use_native_uuid:
-
- def process(value):
- if value is not None:
- value = str(value)
- return value
-
- return process
-
-
class _PsycopgARRAY(PGARRAY):
render_bind_cast = True
{
sqltypes.Numeric: _PsycopgNumeric,
HSTORE: _PsycopgHStore,
- UUID: _PsycopgUUID,
sqltypes.ARRAY: _PsycopgARRAY,
},
)
self,
client_encoding=None,
use_native_hstore=True,
- use_native_uuid=True,
**kwargs,
):
PGDialect.__init__(self, **kwargs)
if not use_native_hstore:
self._has_native_hstore = False
self.use_native_hstore = use_native_hstore
- self.use_native_uuid = use_native_uuid
self.client_encoding = client_encoding
def create_connect_args(self, url):
from .base import PGExecutionContext
from .base import PGIdentifierPreparer
from .base import REGCLASS
-from .base import UUID
from ... import exc
from ... import pool
from ... import util
from ...util.concurrency import await_only
-try:
- from uuid import UUID as _python_UUID # noqa
-except ImportError:
- _python_UUID = None
-
-
class AsyncpgString(sqltypes.String):
render_bind_cast = True
return process
-class AsyncpgUUID(UUID):
- render_bind_cast = True
-
- def bind_processor(self, dialect):
- if not self.as_uuid and dialect.use_native_uuid:
-
- def process(value):
- if value is not None:
- value = _python_UUID(value)
- return value
-
- return process
-
- def result_processor(self, dialect, coltype):
- if not self.as_uuid and dialect.use_native_uuid:
-
- def process(value):
- if value is not None:
- value = str(value)
- return value
-
- return process
-
-
class AsyncpgNumeric(sqltypes.Numeric):
render_bind_cast = True
statement_compiler = PGCompiler_asyncpg
preparer = PGIdentifierPreparer_asyncpg
- use_native_uuid = True
-
colspecs = util.update_copy(
PGDialect.colspecs,
{
sqltypes.DateTime: AsyncpgDateTime,
sqltypes.Interval: AsyncPgInterval,
INTERVAL: AsyncPgInterval,
- UUID: AsyncpgUUID,
sqltypes.Boolean: AsyncpgBoolean,
sqltypes.Integer: AsyncpgInteger,
sqltypes.BigInteger: AsyncpgBigInteger,
import datetime as dt
import re
from typing import Any
-from typing import overload
-from typing import TypeVar
-from uuid import UUID as _python_UUID
from . import array as _array
from . import dml
from ...types import REAL
from ...types import SMALLINT
from ...types import TEXT
+from ...types import UUID as UUID
from ...types import VARCHAR
-from ...util.typing import Literal
IDX_USING = re.compile(r"^(?:btree|hash|gist|gin|[\w_]+)$", re.I)
_INT_TYPES = (20, 21, 23, 26, 1005, 1007, 1016)
+class PGUuid(UUID):
+ render_bind_cast = True
+ render_literal_cast = True
+
+
class BYTEA(sqltypes.LargeBinary[bytes]):
__visit_name__ = "BYTEA"
PGBit = BIT
-_UUID_RETURN = TypeVar("_UUID_RETURN", str, _python_UUID)
-
-
-class UUID(sqltypes.TypeEngine[_UUID_RETURN]):
-
- """PostgreSQL UUID type.
-
- Represents the UUID column type, interpreting
- data either as natively returned by the DBAPI
- or as Python uuid objects.
-
- The UUID type is currently known to work within the prominent DBAPI
- drivers supported by SQLAlchemy including psycopg, psycopg2, pg8000 and
- asyncpg. Support for other DBAPI drivers may be incomplete or non-present.
-
- """
-
- __visit_name__ = "UUID"
-
- @overload
- def __init__(self: "UUID[_python_UUID]", as_uuid: Literal[True] = ...):
- ...
-
- @overload
- def __init__(self: "UUID[str]", as_uuid: Literal[False] = ...):
- ...
-
- def __init__(self, as_uuid: bool = True):
- """Construct a UUID type.
-
-
- :param as_uuid=True: if True, values will be interpreted
- as Python uuid objects, converting to/from string via the
- DBAPI.
-
- .. versionchanged: 2 ``as_uuid`` now defaults to ``True``.
-
- """
- self.as_uuid = as_uuid
-
- def coerce_compared_value(self, op, value):
- """See :meth:`.TypeEngine.coerce_compared_value` for a description."""
-
- if isinstance(value, str):
- return self
- else:
- return super(UUID, self).coerce_compared_value(op, value)
-
- def bind_processor(self, dialect):
- if self.as_uuid:
-
- def process(value):
- if value is not None:
- value = str(value)
- return value
-
- return process
- else:
- return None
-
- def result_processor(self, dialect, coltype):
- if self.as_uuid:
-
- def process(value):
- if value is not None:
- value = _python_UUID(value)
- return value
-
- return process
- else:
- return None
-
- def literal_processor(self, dialect):
- if self.as_uuid:
-
- def process(value):
- if value is not None:
- value = "'%s'::UUID" % value
- return value
-
- return process
- else:
-
- def process(value):
- if value is not None:
- value = "'%s'" % value
- return value
-
- return process
-
- @property
- def python_type(self):
- return _python_UUID if self.as_uuid else str
-
-
-PGUuid = UUID
-
class TSVECTOR(sqltypes.TypeEngine[Any]):
sqltypes.Enum: ENUM,
sqltypes.JSON.JSONPathType: _json.JSONPathType,
sqltypes.JSON: _json.JSON,
+ UUID: PGUuid,
}
ischema_names = {
compiled = "BIT(%d)" % type_.length
return compiled
+ def visit_uuid(self, type_, **kw):
+ if type_.native_uuid:
+ return self.visit_UUID(type_, **kw)
+ else:
+ return super().visit_uuid(type_, **kw)
+
def visit_UUID(self, type_, **kw):
return "UUID"
supports_native_enum = True
supports_native_boolean = True
+ supports_native_uuid = True
supports_smallserial = True
supports_sequences = True
""" # noqa
import decimal
import re
-from uuid import UUID as _python_UUID
from .array import ARRAY as PGARRAY
from .base import _DECIMAL_TYPES
from .base import PGDialect
from .base import PGExecutionContext
from .base import PGIdentifierPreparer
-from .base import UUID
from .json import JSON
from .json import JSONB
from .json import JSONPathType
# DBAPI type 1009
-class _PGUUID(UUID):
- render_bind_cast = True
-
- def bind_processor(self, dialect):
- if not self.as_uuid:
-
- def process(value):
- if value is not None:
- value = _python_UUID(value)
- return value
-
- return process
-
- def result_processor(self, dialect, coltype):
- if not self.as_uuid:
-
- def process(value):
- if value is not None:
- value = str(value)
- return value
-
- return process
-
-
class _PGEnum(ENUM):
def get_dbapi_type(self, dbapi):
return dbapi.UNKNOWN
sqltypes.JSON.JSONIndexType: _PGJSONIndexType,
sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType,
sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType,
- UUID: _PGUUID,
sqltypes.Interval: _PGInterval,
INTERVAL: _PGInterval,
sqltypes.DateTime: _PGTimeStamp,
from ._psycopg_common import _PGDialect_common_psycopg
from ._psycopg_common import _PGExecutionContext_common_psycopg
-from ._psycopg_common import _PsycopgUUID
from .base import INTERVAL
from .base import PGCompiler
from .base import PGIdentifierPreparer
-from .base import UUID
from .json import JSON
from .json import JSONB
from .json import JSONPathType
pass
-class _PGUUID(_PsycopgUUID):
- render_bind_cast = True
-
-
class _PGInterval(INTERVAL):
render_bind_cast = True
sqltypes.JSON.JSONPathType: _PGJSONPathType,
sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType,
sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType,
- UUID: _PGUUID,
sqltypes.Interval: _PGInterval,
INTERVAL: _PGInterval,
sqltypes.Date: _PGDate,
fns.append(on_connect)
- if self.dbapi and self.use_native_uuid:
+ if self.dbapi:
def on_connect(dbapi_conn):
extras.register_uuid(None, dbapi_conn)
supports_native_enum = False
supports_native_boolean = False
+ supports_native_uuid = False
non_native_boolean_check_constraint = True
supports_simple_order_by_label = True
"""indicates if Decimal objects are handled and returned for precision
numeric types, or if floats are returned"""
+ supports_native_uuid: bool
+ """indicates if Python UUID() objects are handled natively by the
+ driver for SQL UUID datatypes.
+
+ .. versionadded:: 2.0
+
+ """
+
construct_arguments: Optional[
List[Tuple[Type[ClauseElement], Mapping[str, Any]]]
] = None
def bindparam(
key: Optional[str],
value: Any = _NoArg.NO_ARG,
- type_: Optional[TypeEngine[_T]] = None,
+ type_: Optional[_TypeEngineArgument[_T]] = None,
unique: bool = False,
required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
quote: Optional[bool] = None,
_resolve_literal_only = True
def _implicit_coercions(
- self, element, resolved, argname, type_=None, **kw
+ self,
+ element,
+ resolved,
+ argname,
+ type_=None,
+ literal_execute=False,
+ **kw,
):
if not _is_literal(resolved):
self._raise_for_expected(
element, resolved=resolved, argname=argname, **kw
)
- return elements.BindParameter(None, element, type_=type_, unique=True)
+ return elements.BindParameter(
+ None,
+ element,
+ type_=type_,
+ unique=True,
+ literal_execute=literal_execute,
+ )
def _literal_coercion(self, element, argname=None, type_=None, **kw):
return element
def bindparam_string(
self,
- name,
- positional_names=None,
- post_compile=False,
- expanding=False,
- escaped_from=None,
- bindparam_type=None,
- **kw,
- ):
+ name: str,
+ positional_names: Optional[List[str]] = None,
+ post_compile: bool = False,
+ expanding: bool = False,
+ escaped_from: Optional[str] = None,
+ bindparam_type: Optional[TypeEngine[Any]] = None,
+ **kw: Any,
+ ) -> str:
if self.positional:
if positional_names is not None:
{escaped_from: name}
)
if post_compile:
- return "__[POSTCOMPILE_%s]" % name
-
- ret = self.bindtemplate % {"name": name}
+ ret = "__[POSTCOMPILE_%s]" % name
+ if expanding:
+ # for expanding, bound parameters or literal values will be
+ # rendered per item
+ return ret
+
+ # otherwise, for non-expanding "literal execute", apply
+ # bind casts as determined by the datatype
+ if bindparam_type is not None:
+ type_impl = bindparam_type._unwrapped_dialect_impl(
+ self.dialect
+ )
+ if type_impl.render_literal_cast:
+ ret = self.render_bind_cast(bindparam_type, type_impl, ret)
+ return ret
+ else:
+ ret = self.bindtemplate % {"name": name}
if (
bindparam_type is not None
def visit_NCLOB(self, type_, **kw):
return "NCLOB"
- def _render_string_type(self, type_, name):
+ def _render_string_type(self, type_, name, length_override=None):
text = name
- if type_.length:
+ if length_override:
+ text += "(%d)" % length_override
+ elif type_.length:
text += "(%d)" % type_.length
if type_.collation:
text += ' COLLATE "%s"' % type_.collation
def visit_BOOLEAN(self, type_, **kw):
return "BOOLEAN"
+ def visit_uuid(self, type_, **kw):
+ return self._render_string_type(type_, "CHAR", length_override=32)
+
def visit_large_binary(self, type_, **kw):
return self.visit_BLOB(type_, **kw)
def literal(
- value: Any, type_: Optional[_TypeEngineArgument[_T]] = None
+ value: Any,
+ type_: Optional[_TypeEngineArgument[_T]] = None,
+ literal_execute: bool = False,
) -> BindParameter[_T]:
r"""Return a literal clause, bound to a bind parameter.
:class:`BindParameter` with a bound value.
:param value: the value to be bound. Can be any Python object supported by
- the underlying DB-API, or is translatable via the given type argument.
+ the underlying DB-API, or is translatable via the given type argument.
- :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which
- will provide bind-parameter translation for this literal.
+ :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
+ provide bind-parameter translation for this literal.
+
+ :param literal_execute: optional bool, when True, the SQL engine will
+ attempt to render the bound value directly in the SQL statement at
+ execution time rather than providing as a parameter value.
+
+ .. versionadded:: 2.0
"""
- return coercions.expect(roles.LiteralValueRole, value, type_=type_)
+ return coercions.expect(
+ roles.LiteralValueRole,
+ value,
+ type_=type_,
+ literal_execute=literal_execute,
+ )
def literal_column(
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
+from uuid import UUID as _python_UUID
from . import coercions
from . import elements
class _RenderISO8601NoT:
- def literal_processor(self, dialect):
- def process(value):
- if value is not None:
- value = f"""'{value.isoformat().replace("T", " ")}'"""
- return value
+ def _literal_processor_datetime(self, dialect):
+ return self._literal_processor_portion(dialect, None)
+
+ def _literal_processor_date(self, dialect):
+ return self._literal_processor_portion(dialect, 0)
+
+ def _literal_processor_time(self, dialect):
+ return self._literal_processor_portion(dialect, -1)
+
+ def _literal_processor_portion(self, dialect, _portion=None):
+ assert _portion in (None, 0, -1)
+ if _portion is not None:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{value.isoformat().split("T")[_portion]}'"""
+ return value
+
+ else:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{value.isoformat().replace("T", " ")}'"""
+ return value
return process
else:
return self
+ def literal_processor(self, dialect):
+ return self._literal_processor_datetime(dialect)
+
@property
def python_type(self):
return dt.datetime
def python_type(self):
return dt.date
+ def literal_processor(self, dialect):
+ return self._literal_processor_date(dialect)
+
@util.memoized_property
def _expression_adaptations(self):
# Based on https://www.postgresql.org/docs/current/\
operators.sub: {Time: Interval, Interval: self.__class__},
}
+ def literal_processor(self, dialect):
+ return self._literal_processor_time(dialect)
+
class _Binary(TypeEngine[bytes]):
"""
+_UUID_RETURN = TypeVar("_UUID_RETURN", str, _python_UUID)
+
+
+class Uuid(TypeEngine[_UUID_RETURN]):
+
+ """Represent a database agnostic UUID datatype.
+
+ For backends that have no "native" UUID datatype, the value will
+ make use of ``CHAR(32)`` and store the UUID as a 32-character alphanumeric
+ hex string.
+
+ For backends which are known to support ``UUID`` directly or a similar
+ uuid-storing datatype such as SQL Server's ``UNIQUEIDENTIFIER``, a
+ "native" mode enabled by default allows these types will be used on those
+ backends.
+
+ .. versionadded:: 2.0
+
+ .. seealso::
+
+ :class:`_sqltypes.UUID` - represents exactly the ``UUID`` datatype
+ without any backend-agnostic behaviors.
+
+ """
+
+ __visit_name__ = "uuid"
+
+ collation = None
+
+ @overload
+ def __init__(
+ self: "Uuid[_python_UUID]",
+ as_uuid: Literal[True] = ...,
+ native_uuid: bool = ...,
+ ):
+ ...
+
+ @overload
+ def __init__(
+ self: "Uuid[str]",
+ as_uuid: Literal[False] = ...,
+ native_uuid: bool = ...,
+ ):
+ ...
+
+ def __init__(self, as_uuid: bool = True, native_uuid: bool = True):
+ """Construct a :class:`_sqltypes.Uuid` type.
+
+ :param as_uuid=True: if True, values will be interpreted
+ as Python uuid objects, converting to/from string via the
+ DBAPI.
+
+ .. versionchanged: 2.0 ``as_uuid`` now defaults to ``True``.
+
+ :param native_uuid=True: if True, backends that support either the
+ ``UUID`` datatype directly, or a UUID-storing value
+ (such as SQL Server's ``UNIQUEIDENTIFIER`` will be used by those
+ backends. If False, a ``CHAR(32)`` datatype will be used for
+ all backends regardless of native support.
+
+ """
+ self.as_uuid = as_uuid
+ self.native_uuid = native_uuid
+
+ @property
+ def python_type(self):
+ return _python_UUID if self.as_uuid else str
+
+ def coerce_compared_value(self, op, value):
+ """See :meth:`.TypeEngine.coerce_compared_value` for a description."""
+
+ if isinstance(value, str):
+ return self
+ else:
+ return super().coerce_compared_value(op, value)
+
+ def bind_processor(self, dialect):
+ character_based_uuid = (
+ not dialect.supports_native_uuid or not self.native_uuid
+ )
+
+ if character_based_uuid:
+ if self.as_uuid:
+
+ def process(value):
+ if value is not None:
+ value = value.hex
+ return value
+
+ return process
+ else:
+
+ def process(value):
+ if value is not None:
+ value = value.replace("-", "")
+ return value
+
+ return process
+ else:
+ return None
+
+ def result_processor(self, dialect, coltype):
+ character_based_uuid = (
+ not dialect.supports_native_uuid or not self.native_uuid
+ )
+
+ if character_based_uuid:
+ if self.as_uuid:
+
+ def process(value):
+ if value is not None:
+ value = _python_UUID(value)
+ return value
+
+ return process
+ else:
+
+ def process(value):
+ if value is not None:
+ value = str(_python_UUID(value))
+ return value
+
+ return process
+ else:
+
+ if not self.as_uuid:
+
+ def process(value):
+ if value is not None:
+ value = str(value)
+ return value
+
+ return process
+ else:
+ return None
+
+ def literal_processor(self, dialect):
+ character_based_uuid = (
+ not dialect.supports_native_uuid or not self.native_uuid
+ )
+
+ if not self.as_uuid:
+
+ def process(value):
+ if value is not None:
+ value = (
+ f"""'{value.replace("-", "").replace("'", "''")}'"""
+ )
+ return value
+
+ return process
+ else:
+ if character_based_uuid:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{value.hex}'"""
+ return value
+
+ return process
+ else:
+
+ def process(value):
+ if value is not None:
+ value = f"""'{str(value).replace("'", "''")}'"""
+ return value
+
+ return process
+
+
+class UUID(Uuid[_UUID_RETURN]):
+
+ """Represent the SQL UUID type.
+
+ This is the SQL-native form of the :class:`_types.Uuid` database agnostic
+ datatype, and is backwards compatible with the previous PostgreSQL-only
+ version of ``UUID``.
+
+ The :class:`_sqltypes.UUID` datatype only works on databases that have a
+ SQL datatype named ``UUID``. It will not function for backends which don't
+ have this exact-named type, including SQL Server. For backend-agnostic UUID
+ values with native support, including for SQL Server's ``UNIQUEIDENTIFIER``
+ datatype, use the :class:`_sqltypes.Uuid` datatype.
+
+ .. versionadded:: 2.0
+
+ .. seealso::
+
+ :class:`_sqltypes.Uuid`
+
+ """
+
+ __visit_name__ = "UUID"
+
+ @overload
+ def __init__(self: "UUID[_python_UUID]", as_uuid: Literal[True] = ...):
+ ...
+
+ @overload
+ def __init__(self: "UUID[str]", as_uuid: Literal[False] = ...):
+ ...
+
+ def __init__(self, as_uuid: bool = True):
+ """Construct a :class:`_sqltypes.UUID` type.
+
+
+ :param as_uuid=True: if True, values will be interpreted
+ as Python uuid objects, converting to/from string via the
+ DBAPI.
+
+ .. versionchanged: 2.0 ``as_uuid`` now defaults to ``True``.
+
+ """
+ self.as_uuid = as_uuid
+ self.native_uuid = True
+
+
NULLTYPE = NullType()
BOOLEANTYPE = Boolean()
STRINGTYPE = String()
int: Integer(),
float: Float(),
bool: BOOLEANTYPE,
+ _python_UUID: Uuid(),
decimal.Decimal: Numeric(),
dt.date: Date(),
dt.datetime: _DATETIME,
"""
+ render_literal_cast = False
+ """render casts when rendering a value as an inline literal,
+ e.g. with :meth:`.TypeEngine.literal_processor`.
+
+ .. versionadded:: 2.0
+
+ """
+
class Comparator(
ColumnOperators,
Generic[_CT],
return exclusions.open()
+ @property
+ def unicode_data_no_special_types(self):
+ """Target database/dialect can receive / deliver / compare data with
+ non-ASCII characters in plain VARCHAR, TEXT columns, without the need
+ for special "national" datatypes like NVARCHAR or similar.
+
+ """
+ return exclusions.open()
+
@property
def unicode_data(self):
"""Target database/dialect must support Python unicode objects with
if "test_needs_fk" in test_opts or "test_needs_acid" in test_opts:
kw["mysql_engine"] = "InnoDB"
else:
+ # there are in fact test fixtures that rely upon MyISAM,
+ # due to MySQL / MariaDB having poor FK behavior under innodb,
+ # such as a self-referential table can't be deleted from at
+ # once without attending to per-row dependencies. We'd need to
+ # add special steps to some fixtures if we want to not
+ # explicitly state MyISAM here
kw["mysql_engine"] = "MyISAM"
elif exclusions.against(config._current, "mariadb"):
if (
import decimal
import json
import re
+import uuid
from .. import config
from .. import engines
from ... import TypeDecorator
from ... import Unicode
from ... import UnicodeText
+from ... import UUID
+from ... import Uuid
from ...orm import declarative_base
from ...orm import Session
from ...sql.sqltypes import LargeBinary
# official type; ideally we'd be able to use CAST here
# but MySQL in particular can't CAST fully
- def run(type_, input_, output, filter_=None):
+ def run(
+ type_,
+ input_,
+ output,
+ filter_=None,
+ compare=None,
+ support_whereclause=True,
+ ):
t = Table("t", metadata, Column("x", type_))
t.create(connection)
for value in input_:
- ins = (
- t.insert()
- .values(x=literal(value, type_))
- .compile(
- dialect=testing.db.dialect,
- compile_kwargs=dict(literal_binds=True),
- )
+ ins = t.insert().values(
+ x=literal(value, type_, literal_execute=True)
)
connection.execute(ins)
- if self.supports_whereclause:
- stmt = t.select().where(t.c.x == literal(value))
+ if support_whereclause and self.supports_whereclause:
+ if compare:
+ stmt = t.select().where(
+ t.c.x
+ == literal(
+ compare,
+ type_,
+ literal_execute=True,
+ ),
+ t.c.x
+ == literal(
+ input_[0],
+ type_,
+ literal_execute=True,
+ ),
+ )
+ else:
+ stmt = t.select().where(
+ t.c.x
+ == literal(
+ compare if compare is not None else input_[0],
+ type_,
+ literal_execute=True,
+ )
+ )
else:
stmt = t.select()
- stmt = stmt.compile(
- dialect=testing.db.dialect,
- compile_kwargs=dict(literal_binds=True),
- )
- for row in connection.execute(stmt):
+ rows = connection.execute(stmt).all()
+ assert rows, "No rows returned"
+ for row in rows:
value = row[0]
if filter_ is not None:
value = filter_(value)
def test_literal(self, literal_round_trip):
literal_round_trip(Text, ["some text"], ["some text"])
+ @requirements.unicode_data_no_special_types
def test_literal_non_ascii(self, literal_round_trip):
literal_round_trip(Text, ["réve🐍 illé"], ["réve🐍 illé"])
# datatype for the literal part because all strings are unicode
literal_round_trip(String(40), ["some text"], ["some text"])
+ @requirements.unicode_data_no_special_types
def test_literal_non_ascii(self, literal_round_trip):
literal_round_trip(String(40), ["réve🐍 illé"], ["réve🐍 illé"])
@testing.requires.datetime_literals
def test_literal(self, literal_round_trip):
compare = self.compare or self.data
- literal_round_trip(self.datatype, [self.data], [compare])
+
+ literal_round_trip(
+ self.datatype, [self.data], [compare], compare=compare
+ )
@testing.requires.standalone_null_binds_whereclause
def test_null_bound_comparison(self):
class DateTimeCoercedToDateTimeTest(_DateFixture, fixtures.TablesTest):
+ """this particular suite is testing that datetime parameters get
+ coerced to dates, which tends to be something DBAPIs do.
+
+ """
+
__requires__ = "date", "date_coerces_from_datetime"
__backend__ = True
datatype = Date
[15.7563, decimal.Decimal("15.7563")],
[15.7563],
filter_=lambda n: n is not None and round(n, 5) or None,
+ support_whereclause=False,
)
@testing.requires.precision_generic_float_type
)
+class UuidTest(_LiteralRoundTripFixture, fixtures.TablesTest):
+ __backend__ = True
+
+ datatype = Uuid
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "uuid_table",
+ metadata,
+ Column(
+ "id", Integer, primary_key=True, test_needs_autoincrement=True
+ ),
+ Column("uuid_data", cls.datatype),
+ Column("uuid_text_data", cls.datatype(as_uuid=False)),
+ Column("uuid_data_nonnative", Uuid(native_uuid=False)),
+ Column(
+ "uuid_text_data_nonnative",
+ Uuid(as_uuid=False, native_uuid=False),
+ ),
+ )
+
+ def test_uuid_round_trip(self, connection):
+ data = uuid.uuid4()
+ uuid_table = self.tables.uuid_table
+
+ connection.execute(
+ uuid_table.insert(),
+ {"id": 1, "uuid_data": data, "uuid_data_nonnative": data},
+ )
+ row = connection.execute(
+ select(
+ uuid_table.c.uuid_data, uuid_table.c.uuid_data_nonnative
+ ).where(
+ uuid_table.c.uuid_data == data,
+ uuid_table.c.uuid_data_nonnative == data,
+ )
+ ).first()
+ eq_(row, (data, data))
+
+ def test_uuid_text_round_trip(self, connection):
+ data = str(uuid.uuid4())
+ uuid_table = self.tables.uuid_table
+
+ connection.execute(
+ uuid_table.insert(),
+ {
+ "id": 1,
+ "uuid_text_data": data,
+ "uuid_text_data_nonnative": data,
+ },
+ )
+ row = connection.execute(
+ select(
+ uuid_table.c.uuid_text_data,
+ uuid_table.c.uuid_text_data_nonnative,
+ ).where(
+ uuid_table.c.uuid_text_data == data,
+ uuid_table.c.uuid_text_data_nonnative == data,
+ )
+ ).first()
+ eq_((row[0].lower(), row[1].lower()), (data, data))
+
+ def test_literal_uuid(self, literal_round_trip):
+ data = uuid.uuid4()
+ literal_round_trip(self.datatype, [data], [data])
+
+ def test_literal_text(self, literal_round_trip):
+ data = str(uuid.uuid4())
+ literal_round_trip(
+ self.datatype(as_uuid=False),
+ [data],
+ [data],
+ filter_=lambda x: x.lower(),
+ )
+
+ def test_literal_nonnative_uuid(self, literal_round_trip):
+ data = uuid.uuid4()
+ literal_round_trip(Uuid(native_uuid=False), [data], [data])
+
+ def test_literal_nonnative_text(self, literal_round_trip):
+ data = str(uuid.uuid4())
+ literal_round_trip(
+ Uuid(as_uuid=False, native_uuid=False),
+ [data],
+ [data],
+ filter_=lambda x: x.lower(),
+ )
+
+
+class NativeUUIDTest(UuidTest):
+ __requires__ = ("uuid_data_type",)
+
+ datatype = UUID
+
+
__all__ = (
"BinaryTest",
"UnicodeVarcharTest",
"DateHistoricTest",
"StringTest",
"BooleanTest",
+ "UuidTest",
+ "NativeUUIDTest",
)
from .sql.sqltypes import TupleType as TupleType
from .sql.sqltypes import Unicode as Unicode
from .sql.sqltypes import UnicodeText as UnicodeText
+from .sql.sqltypes import UUID as UUID
+from .sql.sqltypes import Uuid as Uuid
from .sql.sqltypes import VARBINARY as VARBINARY
from .sql.sqltypes import VARCHAR as VARCHAR
from .sql.type_api import adapt_type as adapt_type
from sqlalchemy.dialects.mssql import base as mssql
from sqlalchemy.dialects.mssql import ROWVERSION
from sqlalchemy.dialects.mssql import TIMESTAMP
+from sqlalchemy.dialects.mssql import UNIQUEIDENTIFIER
from sqlalchemy.dialects.mssql.base import _MSDate
from sqlalchemy.dialects.mssql.base import BIT
from sqlalchemy.dialects.mssql.base import DATETIMEOFFSET
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_not
from sqlalchemy.testing import pickleable
+from sqlalchemy.testing.suite import test_types
from sqlalchemy.util import b
)
+class UniqueIdentifierTest(test_types.UuidTest):
+ __only_on__ = "mssql"
+ __backend__ = True
+ datatype = UNIQUEIDENTIFIER
+
+
class MyPickleType(types.TypeDecorator):
impl = PickleType
cache_ok = True
class UUIDTest(fixtures.TestBase):
- """Test the bind/return values of the UUID type."""
+ """Test postgresql-specific UUID cases.
+
+ See also generic UUID tests in testing/suite/test_types
+
+ """
__only_on__ = "postgresql >= 8.3"
__backend__ = True
from typing import Type
from typing import TypeVar
from typing import Union
+import uuid
from sqlalchemy import BIGINT
from sqlalchemy import Column
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy import testing
+from sqlalchemy import types
from sqlalchemy import VARCHAR
from sqlalchemy.exc import ArgumentError
from sqlalchemy.orm import as_declarative
"SELECT users.data, users.id FROM users",
)
+ @testing.combinations(
+ (str, types.String),
+ (Decimal, types.Numeric),
+ (float, types.Float),
+ (datetime.datetime, types.DateTime),
+ (uuid.UUID, types.Uuid),
+ argnames="pytype,sqltype",
+ )
+ def test_datatype_lookups(self, decl_base, pytype, sqltype):
+ class MyClass(decl_base):
+ __tablename__ = "mytable"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ data: Mapped[pytype]
+
+ assert isinstance(MyClass.__table__.c.data.type, sqltype)
+
class MixinTest(fixtures.TestBase, testing.AssertsCompiledSQL):
__dialect__ = "default"
"""
return exclusions.open()
+ @property
+ def unicode_data_no_special_types(self):
+ """Target database/dialect can receive / deliver / compare data with
+ non-ASCII characters in plain VARCHAR, TEXT columns, without the need
+ for special "national" datatypes like NVARCHAR or similar.
+
+ """
+ return exclusions.fails_on("mssql")
+
@property
def unicode_ddl(self):
"""Target driver must support some degree of non-ascii symbol names."""
return res is not None
return only_on(["mssql"]) + only_if(check)
+
+ @property
+ def uuid_data_type(self):
+ """Return databases that support the UUID datatype."""
+ return only_on(("postgresql >= 8.3", "mariadb >= 10.7.0"))
def load_dialect_impl(self, dialect):
if dialect.name == "postgresql":
- return dialect.type_descriptor(postgresql.UUID())
+ return dialect.type_descriptor(postgresql.INET())
else:
return dialect.type_descriptor(CHAR(32))
t1 = MyType()
d = postgresql.dialect()
assert t1._type_affinity is String
- assert t1.dialect_impl(d)._type_affinity is postgresql.UUID
+ assert t1.dialect_impl(d)._type_affinity is postgresql.INET
class AsGenericTest(fixtures.TestBase):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
