yield result.scalars()
async def run_sync(
- self, fn: Callable[..., Any], *arg: Any, **kw: Any
- ) -> Any:
+ self, fn: Callable[..., _T], *arg: Any, **kw: Any
+ ) -> _T:
"""Invoke the given synchronous (i.e. not async) callable,
passing a synchronous-style :class:`_engine.Connection` as the first
argument.
)
async def run_sync(
- self, fn: Callable[..., Any], *arg: Any, **kw: Any
- ) -> Any:
+ self, fn: Callable[..., _T], *arg: Any, **kw: Any
+ ) -> _T:
"""Invoke the given synchronous (i.e. not async) callable,
passing a synchronous-style :class:`_orm.Session` as the first
argument.
class _HybridDeleterType(Protocol[_T_co]):
- def __call__(self, instance: Any) -> None:
+ def __call__(s, self: Any) -> None:
...
class _HybridExprCallableType(Protocol[_T]):
def __call__(
- self, cls: Any
+ s, cls: Any
) -> Union[_HasClauseElement, SQLColumnExpression[_T]]:
...
from . import exc
from .util.typing import Literal
+from .util.typing import Protocol
_T = TypeVar("_T", bound=Any)
+_TCov = TypeVar("_TCov", bound=Any, covariant=True)
_F = TypeVar("_F", bound=Callable[..., Any])
-_IN = TypeVar("_IN", bound="Inspectable[Any]")
+_IN = TypeVar("_IN", bound=Any)
_registrars: Dict[type, Union[Literal[True], Callable[[Any], Any]]] = {}
__slots__ = ()
+class _InspectableTypeProtocol(Protocol[_TCov]):
+ """a protocol defining a method that's used when a type (ie the class
+ itself) is passed to inspect().
+
+ """
+
+ def _sa_inspect_type(self) -> _TCov:
+ ...
+
+
+class _InspectableProtocol(Protocol[_TCov]):
+ """a protocol defining a method that's used when an instance is
+ passed to inspect().
+
+ """
+
+ def _sa_inspect_instance(self) -> _TCov:
+ ...
+
+
+@overload
+def inspect(
+ subject: Type[_InspectableTypeProtocol[_IN]], raiseerr: bool = True
+) -> _IN:
+ ...
+
+
+@overload
+def inspect(subject: _InspectableProtocol[_IN], raiseerr: bool = True) -> _IN:
+ ...
+
+
@overload
def inspect(subject: Inspectable[_IN], raiseerr: bool = True) -> _IN:
...
from ..util.typing import is_literal
from ..util.typing import is_newtype
from ..util.typing import Literal
+from ..util.typing import Self
if TYPE_CHECKING:
from ._typing import _O
class DeclarativeAttributeIntercept(
- _DynamicAttributesType, inspection.Inspectable[Mapper[Any]]
+ _DynamicAttributesType,
+ # Inspectable is used only by the mypy plugin
+ inspection.Inspectable[Mapper[Any]],
):
"""Metaclass that may be used in conjunction with the
:class:`_orm.DeclarativeBase` class to support addition of class
"""metaclass that includes @dataclass_transforms"""
-class DeclarativeMeta(
- _DynamicAttributesType, inspection.Inspectable[Mapper[Any]]
-):
+class DeclarativeMeta(DeclarativeAttributeIntercept):
metadata: MetaData
registry: RegistryType
class DeclarativeBase(
+ # Inspectable is used only by the mypy plugin
inspection.Inspectable[InstanceState[Any]],
metaclass=DeclarativeAttributeIntercept,
):
"""
if typing.TYPE_CHECKING:
+
+ def _sa_inspect_type(self) -> Mapper[Self]:
+ ...
+
+ def _sa_inspect_instance(self) -> InstanceState[Self]:
+ ...
+
_sa_registry: ClassVar[_RegistryType]
registry: ClassVar[_RegistryType]
__name__: ClassVar[str]
+ # this ideally should be Mapper[Self], but mypy as of 1.4.1 does not
+ # like it, and breaks the declared_attr_one test. Pyright/pylance is
+ # ok with it.
__mapper__: ClassVar[Mapper[Any]]
"""The :class:`_orm.Mapper` object to which a particular class is
mapped.
)
-class DeclarativeBaseNoMeta(inspection.Inspectable[InstanceState[Any]]):
+class DeclarativeBaseNoMeta(
+ # Inspectable is used only by the mypy plugin
+ inspection.Inspectable[InstanceState[Any]]
+):
"""Same as :class:`_orm.DeclarativeBase`, but does not use a metaclass
to intercept new attributes.
"""
+ # this ideally should be Mapper[Self], but mypy as of 1.4.1 does not
+ # like it, and breaks the declared_attr_one test. Pyright/pylance is
+ # ok with it.
__mapper__: ClassVar[Mapper[Any]]
"""The :class:`_orm.Mapper` object to which a particular class is
mapped.
"""
if typing.TYPE_CHECKING:
+
+ def _sa_inspect_type(self) -> Mapper[Self]:
+ ...
+
+ def _sa_inspect_instance(self) -> InstanceState[Self]:
+ ...
+
__tablename__: Any
"""String name to assign to the generated
:class:`_schema.Table` object, if not specified directly via
Literal[False],
str,
_ColumnExpressionArgument[Any],
- Callable[[], Iterable[ColumnElement[Any]]],
+ Callable[[], _ColumnExpressionArgument[Any]],
+ Callable[[], Iterable[_ColumnExpressionArgument[Any]]],
Iterable[Union[str, _ColumnExpressionArgument[Any]]],
]
ORMBackrefArgument = Union[str, Tuple[str, Dict[str, Any]]]
def __getitem__(self, index: Any) -> ColumnElement[Any]:
...
+ @overload
+ def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]:
+ ...
+
+ @overload
+ def __lshift__(self, other: Any) -> ColumnElement[Any]:
+ ...
+
def __lshift__(self, other: Any) -> ColumnElement[Any]:
...
+ @overload
+ def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]:
+ ...
+
+ @overload
+ def __rshift__(self, other: Any) -> ColumnElement[Any]:
+ ...
+
def __rshift__(self, other: Any) -> ColumnElement[Any]:
...
def _name_setter(self, value: str) -> None:
self.first_name = value
+ @name.inplace.deleter
+ def _name_del(self) -> None:
+ self.first_name = ""
+
+ @name.inplace.expression
+ @classmethod
+ def _name_expr(cls) -> ColumnElement[str]:
+ return cls.first_name + "-"
+
class FirstNameLastName(FirstNameOnly):
last_name: Mapped[str]
-"""
-test inspect()
-
-however this is not really working
-
-"""
-from typing import Any
-from typing import Optional
-
-from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import inspect
-from sqlalchemy import Integer
-from sqlalchemy import String
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import DeclarativeBaseNoMeta
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import Mapper
class A(Base):
__tablename__ = "a"
- id = Column(Integer, primary_key=True)
- data = Column(String)
+ id: Mapped[int] = mapped_column(primary_key=True)
+ data: Mapped[str]
+
+
+class BaseNoMeta(DeclarativeBaseNoMeta):
+ pass
+
+
+class B(BaseNoMeta):
+ __tablename__ = "b"
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+ data: Mapped[str]
+
+# EXPECTED_TYPE: Mapper[Any]
+reveal_type(A.__mapper__)
+# EXPECTED_TYPE: Mapper[Any]
+reveal_type(B.__mapper__)
a1 = A(data="d")
+b1 = B(data="d")
e = create_engine("sqlite://")
-# TODO: I can't get these to work, pylance and mypy both don't want
-# to accommodate for different types for the first argument
+insp_a1 = inspect(a1)
-t: Optional[Any] = inspect(a1)
+t: bool = insp_a1.transient
+# EXPECTED_TYPE: InstanceState[A]
+reveal_type(insp_a1)
+# EXPECTED_TYPE: InstanceState[B]
+reveal_type(inspect(b1))
-m: Mapper[Any] = inspect(A)
+m: Mapper[A] = inspect(A)
+# EXPECTED_TYPE: Mapper[A]
+reveal_type(inspect(A))
+# EXPECTED_TYPE: Mapper[B]
+reveal_type(inspect(B))
-inspect(e).get_table_names()
+tables: list[str] = inspect(e).get_table_names()
i: Inspector = inspect(e)
+# EXPECTED_TYPE: Inspector
+reveal_type(inspect(e))
with e.connect() as conn:
from __future__ import annotations
import typing
+from typing import ClassVar
from typing import List
from typing import Optional
from typing import Set
+from typing import TYPE_CHECKING
+from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
+from sqlalchemy import select
+from sqlalchemy import Table
from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import joinedload
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
+from sqlalchemy.orm import registry
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Session
class Base(DeclarativeBase):
# EXPECTED_RE_TYPE: sqlalchemy.orm.attributes.InstrumentedAttribute\[builtins.set\*?\[relationship.Address\]\]
reveal_type(User.addresses_style_two)
+
+
+mapper_registry: registry = registry()
+
+e = create_engine("sqlite:///")
+
+
+@mapper_registry.mapped
+class A:
+ __tablename__ = "a"
+ id: Mapped[int] = mapped_column(primary_key=True)
+ b_id: Mapped[int] = mapped_column(ForeignKey("b.id"))
+ number: Mapped[int] = mapped_column(primary_key=True)
+ number2: Mapped[int] = mapped_column(primary_key=True)
+ if TYPE_CHECKING:
+ __table__: ClassVar[Table]
+
+
+@mapper_registry.mapped
+class B:
+ __tablename__ = "b"
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ # Omit order_by
+ a1: Mapped[list[A]] = relationship("A", uselist=True)
+
+ # All kinds of order_by
+ a2: Mapped[list[A]] = relationship(
+ "A", uselist=True, order_by=(A.id, A.number)
+ )
+ a3: Mapped[list[A]] = relationship(
+ "A", uselist=True, order_by=[A.id, A.number]
+ )
+ a4: Mapped[list[A]] = relationship("A", uselist=True, order_by=A.id)
+ a5: Mapped[list[A]] = relationship(
+ "A", uselist=True, order_by=A.__table__.c.id
+ )
+ a6: Mapped[list[A]] = relationship("A", uselist=True, order_by="A.number")
+
+ # Same kinds but lambda'd
+ a7: Mapped[list[A]] = relationship(
+ "A", uselist=True, order_by=lambda: (A.id, A.number)
+ )
+ a8: Mapped[list[A]] = relationship(
+ "A", uselist=True, order_by=lambda: [A.id, A.number]
+ )
+ a9: Mapped[list[A]] = relationship(
+ "A", uselist=True, order_by=lambda: A.id
+ )
+
+
+mapper_registry.metadata.drop_all(e)
+mapper_registry.metadata.create_all(e)
+
+with Session(e) as s:
+ s.execute(select(B).options(joinedload(B.a1)))
+ s.execute(select(B).options(joinedload(B.a2)))
+ s.execute(select(B).options(joinedload(B.a3)))
+ s.execute(select(B).options(joinedload(B.a4)))
+ s.execute(select(B).options(joinedload(B.a5)))
+ s.execute(select(B).options(joinedload(B.a7)))
+ s.execute(select(B).options(joinedload(B.a8)))
+ s.execute(select(B).options(joinedload(B.a9)))
reveal_type(c2.bitwise_lshift(5))
# EXPECTED_TYPE: BinaryExpression[Any]
reveal_type(c2.bitwise_rshift(5))
- # EXPECTED_TYPE: ColumnElement[Any]
+ # EXPECTED_TYPE: ColumnElement[int]
reveal_type(c2 << 5)
- # EXPECTED_TYPE: ColumnElement[Any]
+ # EXPECTED_TYPE: ColumnElement[int]
reveal_type(c2 >> 5)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
