Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Ana/typed mapped column #2

Closed
wants to merge 3 commits into from
Closed

Ana/typed mapped column #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8253e5d
Make TypeEngine covariant
acanizares Mar 5, 2024
16565df
Return MappedColumn type matching the type engine, default value type…
acanizares Mar 11, 2024
99cac9a
Add and update mapped_column typing tests
acanizares Mar 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 133 additions & 0 deletions lib/sqlalchemy/orm/_orm_constructors.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,139 @@ def contains_alias(alias: Union[Alias, Subquery]) -> AliasOption:
return AliasOption(alias)


# nullable=True -> MappedColumn[Optional[_T]]
@overload
def mapped_column(
__name_pos: Optional[
Union[str, _TypeEngineArgument[_T], SchemaEventTarget]
] = None,
__type_pos: Optional[
Union[_TypeEngineArgument[_T], SchemaEventTarget]
] = None,
*args: SchemaEventTarget,
init: Union[_NoArg, bool] = _NoArg.NO_ARG,
repr: Union[_NoArg, bool] = _NoArg.NO_ARG, # noqa: A002
default: Optional[Union[_NoArg, _T, Callable[..., _T]]] = _NoArg.NO_ARG,
default_factory: Union[_NoArg, Callable[[], _T]] = _NoArg.NO_ARG,
compare: Union[_NoArg, bool] = _NoArg.NO_ARG,
kw_only: Union[_NoArg, bool] = _NoArg.NO_ARG,
nullable: Literal[True],
primary_key: Optional[bool] = False,
deferred: Union[_NoArg, bool] = _NoArg.NO_ARG,
deferred_group: Optional[str] = None,
deferred_raiseload: Optional[bool] = None,
use_existing_column: bool = False,
name: Optional[str] = None,
type_: Optional[_TypeEngineArgument[Any]] = None,
autoincrement: _AutoIncrementType = "auto",
doc: Optional[str] = None,
key: Optional[str] = None,
index: Optional[bool] = None,
unique: Optional[bool] = None,
info: Optional[_InfoType] = None,
onupdate: Optional[Any] = None,
insert_default: Optional[Any] = _NoArg.NO_ARG,
server_default: Optional[_ServerDefaultArgument] = None,
server_onupdate: Optional[FetchedValue] = None,
active_history: bool = False,
quote: Optional[bool] = None,
system: bool = False,
comment: Optional[str] = None,
sort_order: Union[_NoArg, int] = _NoArg.NO_ARG,
**kw: Any,
) -> MappedColumn[Optional[_T]]: ...


# nullable=False -> MappedColumn[_T]
@overload
def mapped_column(
__name_pos: Optional[
Union[str, _TypeEngineArgument[_T], SchemaEventTarget]
] = None,
__type_pos: Optional[
Union[_TypeEngineArgument[_T], SchemaEventTarget]
] = None,
*args: SchemaEventTarget,
init: Union[_NoArg, bool] = _NoArg.NO_ARG,
repr: Union[_NoArg, bool] = _NoArg.NO_ARG, # noqa: A002
default: Union[_NoArg, _T, Callable[..., _T]] = _NoArg.NO_ARG,
default_factory: Union[_NoArg, Callable[[], _T]] = _NoArg.NO_ARG,
compare: Union[_NoArg, bool] = _NoArg.NO_ARG,
kw_only: Union[_NoArg, bool] = _NoArg.NO_ARG,
nullable: Literal[False],
primary_key: Optional[bool] = False,
deferred: Union[_NoArg, bool] = _NoArg.NO_ARG,
deferred_group: Optional[str] = None,
deferred_raiseload: Optional[bool] = None,
use_existing_column: bool = False,
name: Optional[str] = None,
type_: Optional[_TypeEngineArgument[Any]] = None,
autoincrement: _AutoIncrementType = "auto",
doc: Optional[str] = None,
key: Optional[str] = None,
index: Optional[bool] = None,
unique: Optional[bool] = None,
info: Optional[_InfoType] = None,
onupdate: Optional[Any] = None,
insert_default: Optional[Any] = _NoArg.NO_ARG,
server_default: Optional[_ServerDefaultArgument] = None,
server_onupdate: Optional[FetchedValue] = None,
active_history: bool = False,
quote: Optional[bool] = None,
system: bool = False,
comment: Optional[str] = None,
sort_order: Union[_NoArg, int] = _NoArg.NO_ARG,
**kw: Any,
) -> MappedColumn[_T]: ...


# nullable unset or None -> MappedColumn[_T]
# TODO this would be only correct if the default unset nullable was False,
# which implies a larger change.
@overload
def mapped_column(
__name_pos: Optional[
Union[str, _TypeEngineArgument[_T], SchemaEventTarget]
] = None,
__type_pos: Optional[
Union[_TypeEngineArgument[_T], SchemaEventTarget]
] = None,
*args: SchemaEventTarget,
init: Union[_NoArg, bool] = _NoArg.NO_ARG,
repr: Union[_NoArg, bool] = _NoArg.NO_ARG, # noqa: A002
default: Union[_NoArg, _T, Callable[..., _T]] = _NoArg.NO_ARG,
default_factory: Union[_NoArg, Callable[[], _T]] = _NoArg.NO_ARG,
compare: Union[_NoArg, bool] = _NoArg.NO_ARG,
kw_only: Union[_NoArg, bool] = _NoArg.NO_ARG,
nullable: Optional[
Literal[SchemaConst.NULL_UNSPECIFIED]
] = SchemaConst.NULL_UNSPECIFIED,
primary_key: Optional[bool] = False,
deferred: Union[_NoArg, bool] = _NoArg.NO_ARG,
deferred_group: Optional[str] = None,
deferred_raiseload: Optional[bool] = None,
use_existing_column: bool = False,
name: Optional[str] = None,
type_: Optional[_TypeEngineArgument[Any]] = None,
autoincrement: _AutoIncrementType = "auto",
doc: Optional[str] = None,
key: Optional[str] = None,
index: Optional[bool] = None,
unique: Optional[bool] = None,
info: Optional[_InfoType] = None,
onupdate: Optional[Any] = None,
insert_default: Optional[Any] = _NoArg.NO_ARG,
server_default: Optional[_ServerDefaultArgument] = None,
server_onupdate: Optional[FetchedValue] = None,
active_history: bool = False,
quote: Optional[bool] = None,
system: bool = False,
comment: Optional[str] = None,
sort_order: Union[_NoArg, int] = _NoArg.NO_ARG,
**kw: Any,
) -> MappedColumn[_T]: ...


def mapped_column(
__name_pos: Optional[
Union[str, _TypeEngineArgument[Any], SchemaEventTarget]
Expand Down
40 changes: 20 additions & 20 deletions lib/sqlalchemy/sql/type_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ def __call__(
) -> TypeEngine.Comparator[_T]: ...


class TypeEngine(Visitable, Generic[_T]):
class TypeEngine(Visitable, Generic[_T_co]):
"""The ultimate base class for all SQL datatypes.

Common subclasses of :class:`.TypeEngine` include
Expand Down Expand Up @@ -359,7 +359,7 @@ def copy_value(self, value: Any) -> Any:

def literal_processor(
self, dialect: Dialect
) -> Optional[_LiteralProcessorType[_T]]:
) -> Optional[_LiteralProcessorType[_T_co]]:
"""Return a conversion function for processing literal values that are
to be rendered directly without using binds.

Expand Down Expand Up @@ -396,7 +396,7 @@ class explicitly.

def bind_processor(
self, dialect: Dialect
) -> Optional[_BindProcessorType[_T]]:
) -> Optional[_BindProcessorType[_T_co]]:
"""Return a conversion function for processing bind values.

Returns a callable which will receive a bind parameter value
Expand Down Expand Up @@ -432,7 +432,7 @@ class explicitly.

def result_processor(
self, dialect: Dialect, coltype: object
) -> Optional[_ResultProcessorType[_T]]:
) -> Optional[_ResultProcessorType[_T_co]]:
"""Return a conversion function for processing result row values.

Returns a callable which will receive a result row column
Expand Down Expand Up @@ -468,8 +468,8 @@ class explicitly.
return None

def column_expression(
self, colexpr: ColumnElement[_T]
) -> Optional[ColumnElement[_T]]:
self, colexpr: ColumnElement[_T_co]
) -> Optional[ColumnElement[_T_co]]:
"""Given a SELECT column expression, return a wrapping SQL expression.

This is typically a SQL function that wraps a column expression
Expand Down Expand Up @@ -527,8 +527,8 @@ def _has_column_expression(self) -> bool:
)

def bind_expression(
self, bindvalue: BindParameter[_T]
) -> Optional[ColumnElement[_T]]:
self, bindvalue: BindParameter[_T_co]
) -> Optional[ColumnElement[_T_co]]:
"""Given a bind value (i.e. a :class:`.BindParameter` instance),
return a SQL expression in its place.

Expand Down Expand Up @@ -576,7 +576,7 @@ class explicitly.

def _sentinel_value_resolver(
self, dialect: Dialect
) -> Optional[_SentinelProcessorType[_T]]:
) -> Optional[_SentinelProcessorType[_T_co]]:
"""Return an optional callable that will match parameter values
(post-bind processing) to result values
(pre-result-processing), for use in the "sentinel" feature.
Expand Down Expand Up @@ -768,7 +768,7 @@ def _resolve_for_python_type(
return self

@util.ro_memoized_property
def _type_affinity(self) -> Optional[Type[TypeEngine[_T]]]:
def _type_affinity(self) -> Optional[Type[TypeEngine[_T_co]]]:
"""Return a rudimental 'affinity' value expressing the general class
of type."""

Expand All @@ -784,7 +784,7 @@ def _type_affinity(self) -> Optional[Type[TypeEngine[_T]]]:
@util.ro_memoized_property
def _generic_type_affinity(
self,
) -> Type[TypeEngine[_T]]:
) -> Type[TypeEngine[_T_co]]:
best_camelcase = None
best_uppercase = None

Expand All @@ -811,10 +811,10 @@ def _generic_type_affinity(
return (
best_camelcase
or best_uppercase
or cast("Type[TypeEngine[_T]]", NULLTYPE.__class__)
or cast("Type[TypeEngine[_T_co]]", NULLTYPE.__class__)
)

def as_generic(self, allow_nulltype: bool = False) -> TypeEngine[_T]:
def as_generic(self, allow_nulltype: bool = False) -> TypeEngine[_T_co]:
"""
Return an instance of the generic type corresponding to this type
using heuristic rule. The method may be overridden if this
Expand Down Expand Up @@ -854,7 +854,7 @@ def as_generic(self, allow_nulltype: bool = False) -> TypeEngine[_T]:

return util.constructor_copy(self, self._generic_type_affinity)

def dialect_impl(self, dialect: Dialect) -> TypeEngine[_T]:
def dialect_impl(self, dialect: Dialect) -> TypeEngine[_T_co]:
"""Return a dialect-specific implementation for this
:class:`.TypeEngine`.

Expand All @@ -867,7 +867,7 @@ def dialect_impl(self, dialect: Dialect) -> TypeEngine[_T]:
return tm["impl"]
return self._dialect_info(dialect)["impl"]

def _unwrapped_dialect_impl(self, dialect: Dialect) -> TypeEngine[_T]:
def _unwrapped_dialect_impl(self, dialect: Dialect) -> TypeEngine[_T_co]:
"""Return the 'unwrapped' dialect impl for this type.

For a type that applies wrapping logic (e.g. TypeDecorator), give
Expand All @@ -883,7 +883,7 @@ def _unwrapped_dialect_impl(self, dialect: Dialect) -> TypeEngine[_T]:

def _cached_literal_processor(
self, dialect: Dialect
) -> Optional[_LiteralProcessorType[_T]]:
) -> Optional[_LiteralProcessorType[_T_co]]:
"""Return a dialect-specific literal processor for this type."""

try:
Expand All @@ -899,7 +899,7 @@ def _cached_literal_processor(

def _cached_bind_processor(
self, dialect: Dialect
) -> Optional[_BindProcessorType[_T]]:
) -> Optional[_BindProcessorType[_T_co]]:
"""Return a dialect-specific bind processor for this type."""

try:
Expand All @@ -915,7 +915,7 @@ def _cached_bind_processor(

def _cached_result_processor(
self, dialect: Dialect, coltype: Any
) -> Optional[_ResultProcessorType[_T]]:
) -> Optional[_ResultProcessorType[_T_co]]:
"""Return a dialect-specific result processor for this type."""

try:
Expand All @@ -935,7 +935,7 @@ def _cached_result_processor(

def _cached_sentinel_value_processor(
self, dialect: Dialect
) -> Optional[_SentinelProcessorType[_T]]:
) -> Optional[_SentinelProcessorType[_T_co]]:
try:
return dialect._type_memos[self]["sentinel"]
except KeyError:
Expand All @@ -946,7 +946,7 @@ def _cached_sentinel_value_processor(
return bp

def _cached_custom_processor(
self, dialect: Dialect, key: str, fn: Callable[[TypeEngine[_T]], _O]
self, dialect: Dialect, key: str, fn: Callable[[TypeEngine[_T_co]], _O]
) -> _O:
"""return a dialect-specific processing object for
custom purposes.
Expand Down
3 changes: 2 additions & 1 deletion test/typing/plain_files/orm/issue_9340.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from typing import Optional
from typing import Sequence
from typing import TYPE_CHECKING

Expand Down Expand Up @@ -28,7 +29,7 @@ class UserComment(Message):
__mapper_args__ = {
"polymorphic_identity": "user_comment",
}
username: Mapped[str] = mapped_column(nullable=True)
username: Mapped[Optional[str]] = mapped_column(nullable=True)


engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/")
Expand Down
Loading