mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-06-20 23:52:05 -04:00
Fix Session bulk mappings typing for mapped classes
Fixes #9256.
This updates the annotations for Session.bulk_insert_mappings() and Session.bulk_update_mappings().
The docstrings and runtime behavior already allow either a mapped class or a Mapper object, but the previous annotations only accepted Mapper[Any].
This patch switches those arguments to the existing _EntityBindKey alias, which matches the inputs accepted by _class_to_mapper(): mapped classes and Mapper objects, but not AliasedClass or AliasedInsp.
I also updated the internal _bulk_save_mappings() annotation so the public methods and the private helper stay consistent. The scoped_session proxy output has been kept in sync with tools/generate_proxy_methods.py, and the generator check passes.
I added a typing regression test covering both mapped classes and Mapper objects for the two bulk mapping methods. I confirmed that the mapped-class cases fail with the old annotation and pass with this change.
Checked locally:
python -m pytest -m mypy test/typing/test_mypy.py -k "session.py" -q
python -m mypy ./lib/sqlalchemy
python tools/generate_proxy_methods.py --check
python -m pytest test/orm/dml/test_bulk.py -q
python -m pytest -m mypy test/typing/test_mypy.py -k "not typed_queries.py" -q
I could not run the full typing suite locally because my local Python 3.12 environment does not include string.templatelib. I only skipped typed_queries.py; that file is expected to be covered by SQLAlchemy's Python 3.14 mypy CI job.
Closes: #13322
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/13322
Pull-request-sha: bb730f3427
Change-Id: I4c5d516b3933b4e7fae9c844881a61f557a8bb5e
This commit is contained in:
committed by
Federico Caselli
parent
092391036c
commit
e00937ec54
@@ -43,7 +43,6 @@ if TYPE_CHECKING:
|
||||
from ._typing import OrmExecuteOptionsParameter
|
||||
from .identity import IdentityMap
|
||||
from .interfaces import ORMOption
|
||||
from .mapper import Mapper
|
||||
from .query import Query
|
||||
from .query import RowReturningQuery
|
||||
from .session import _BindArguments
|
||||
@@ -1392,7 +1391,7 @@ class scoped_session(Generic[_S]):
|
||||
|
||||
def bulk_insert_mappings(
|
||||
self,
|
||||
mapper: Mapper[Any],
|
||||
mapper: _EntityBindKey[Any],
|
||||
mappings: Iterable[Dict[str, Any]],
|
||||
return_defaults: bool = False,
|
||||
render_nulls: bool = False,
|
||||
@@ -1478,7 +1477,7 @@ class scoped_session(Generic[_S]):
|
||||
)
|
||||
|
||||
def bulk_update_mappings(
|
||||
self, mapper: Mapper[Any], mappings: Iterable[Dict[str, Any]]
|
||||
self, mapper: _EntityBindKey[Any], mappings: Iterable[Dict[str, Any]]
|
||||
) -> None:
|
||||
r"""Perform a bulk update of the given list of mapping dictionaries.
|
||||
|
||||
|
||||
@@ -4742,7 +4742,7 @@ class Session(_SessionClassMethods, EventTarget):
|
||||
|
||||
def bulk_insert_mappings(
|
||||
self,
|
||||
mapper: Mapper[Any],
|
||||
mapper: _EntityBindKey[Any],
|
||||
mappings: Iterable[Dict[str, Any]],
|
||||
return_defaults: bool = False,
|
||||
render_nulls: bool = False,
|
||||
@@ -4824,7 +4824,7 @@ class Session(_SessionClassMethods, EventTarget):
|
||||
)
|
||||
|
||||
def bulk_update_mappings(
|
||||
self, mapper: Mapper[Any], mappings: Iterable[Dict[str, Any]]
|
||||
self, mapper: _EntityBindKey[Any], mappings: Iterable[Dict[str, Any]]
|
||||
) -> None:
|
||||
"""Perform a bulk update of the given list of mapping dictionaries.
|
||||
|
||||
@@ -4873,7 +4873,7 @@ class Session(_SessionClassMethods, EventTarget):
|
||||
|
||||
def _bulk_save_mappings(
|
||||
self,
|
||||
mapper: Mapper[_O],
|
||||
mapper: _EntityBindKey[_O],
|
||||
mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
|
||||
*,
|
||||
isupdate: bool,
|
||||
|
||||
@@ -10,6 +10,7 @@ from typing import Unpack
|
||||
from sqlalchemy import Column
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy import ForeignKey
|
||||
from sqlalchemy import inspect
|
||||
from sqlalchemy import Integer
|
||||
from sqlalchemy import MetaData
|
||||
from sqlalchemy import Result
|
||||
@@ -120,6 +121,12 @@ with Session(e) as sess:
|
||||
with sess.begin() as tx:
|
||||
assert_type(tx, SessionTransaction)
|
||||
|
||||
# test #9256
|
||||
sess.bulk_insert_mappings(User, [{"id": 1, "name": "u1"}])
|
||||
sess.bulk_update_mappings(User, [{"id": 1, "name": "u1"}])
|
||||
sess.bulk_insert_mappings(inspect(User), [{"id": 2, "name": "u2"}])
|
||||
sess.bulk_update_mappings(inspect(User), [{"id": 2, "name": "u2"}])
|
||||
|
||||
# more result tests in typed_results.py
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user