From f6f1b4d4d9442d4ede81d1acc8ffe2dae3ac685c Mon Sep 17 00:00:00 2001 From: Ilya Konstantinov Date: Thu, 17 Nov 2022 23:34:04 -0500 Subject: [PATCH] Upgrade to mypy 0.950 (#1116) mypy<0.900 is dragging in typed-ast which no longer seems to build on Python 3.11. In order to unblock the release of 5.3.1, we'll pull up to mypy==0.950 (for assert_type) and drag in some minor (non-breaking) improvements from master. --- docs/release_notes.rst | 12 +- examples/attributes.py | 9 +- examples/indexes.py | 9 +- examples/model.py | 69 +++++---- examples/optimistic_locking.py | 2 +- mypy.ini | 4 + pynamodb/indexes.py | 2 +- pynamodb/settings.py | 2 +- pynamodb/transactions.py | 4 +- requirements-dev.txt | 3 +- tests/conftest.py | 10 -- tests/test_mypy.py | 251 --------------------------------- typing_tests/__init__.py | 0 typing_tests/attributes.py | 72 ++++++++++ typing_tests/models.py | 149 +++++++++++++++++++ typing_tests/transactions.py | 28 ++++ 16 files changed, 319 insertions(+), 307 deletions(-) delete mode 100644 tests/conftest.py delete mode 100644 tests/test_mypy.py create mode 100644 typing_tests/__init__.py create mode 100644 typing_tests/attributes.py create mode 100644 typing_tests/models.py create mode 100644 typing_tests/transactions.py diff --git a/docs/release_notes.rst b/docs/release_notes.rst index 829928277..bcf49fbcf 100644 --- a/docs/release_notes.rst +++ b/docs/release_notes.rst @@ -1,13 +1,17 @@ +.. highlight:: none + Release Notes ============= v5.3.1 ---------- -* Fixed issue introduced in 5.3.0: using TableConnection directly (not through a model) +* Fixed issue introduced in 5.3.0: using :py:class:`~pynamodb.connection.table.TableConnection` directly (not through a model) raised the following exception:: pynamodb.exceptions.TableError: Meta-table for '(table-name)' not initialized +* Fix typing on :py:class:`~pynamodb.transactions.TransactGet` (backport of #1057) + v5.3.0 ---------- @@ -146,7 +150,7 @@ v4.3.3 * Add type stubs for indexing into a ``ListAttribute`` for forming conditional expressions (#774) - :: + .. code-block:: python class MyModel(Model): ... @@ -228,7 +232,9 @@ v4.1.0 This is a backwards compatible, minor release. -* In the Model's Meta, you may now provide an AWS session token, which is mostly useful for assumed roles (#700):: +* In the Model's Meta, you may now provide an AWS session token, which is mostly useful for assumed roles (#700): + + .. code-block:: python sts_client = boto3.client("sts") role_object = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="role_name", DurationSeconds=BOTO3_CLIENT_DURATION) diff --git a/examples/attributes.py b/examples/attributes.py index 761eaf359..b7e1fa414 100644 --- a/examples/attributes.py +++ b/examples/attributes.py @@ -2,7 +2,10 @@ A PynamoDB example using a custom attribute """ import pickle -from pynamodb.attributes import BinaryAttribute, UnicodeAttribute +from typing import Any + +from pynamodb.attributes import Attribute, UnicodeAttribute +from pynamodb.constants import BINARY from pynamodb.models import Model @@ -17,7 +20,9 @@ def __str__(self): return "".format(self.name) -class PickleAttribute(BinaryAttribute): +class PickleAttribute(Attribute[Any]): + attr_type = BINARY + """ This class will serializer/deserialize any picklable Python object. The value will be stored as a binary attribute in DynamoDB. diff --git a/examples/indexes.py b/examples/indexes.py index ae6e5aa82..bc122769b 100644 --- a/examples/indexes.py +++ b/examples/indexes.py @@ -86,16 +86,17 @@ class Meta: player_opponent_index = GamePlayerOpponentIndex() opponent_time_index = GameOpponentTimeIndex() + if not GameModel.exists(): GameModel.create_table(wait=True) # Create an item -item = GameModel('1234', datetime.datetime.utcnow()) -item.winner_id = '5678' -item.save() +game = GameModel('1234', datetime.datetime.utcnow()) +game.winner_id = '5678' +game.save() # Indexes can be queried easily using the index's hash key -for item in GameModel.player_opponent_index.query('1234'): +for game in GameModel.player_opponent_index.query('1234'): print("Item queried from index: {0}".format(item)) # Count on an index diff --git a/examples/model.py b/examples/model.py index 8f9e8b1a3..303714ec2 100644 --- a/examples/model.py +++ b/examples/model.py @@ -4,9 +4,15 @@ http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/SampleTablesAndData.html """ import logging +from typing import Any + from pynamodb.models import Model from pynamodb.attributes import ( - UnicodeAttribute, NumberAttribute, UnicodeSetAttribute, UTCDateTimeAttribute + ListAttribute, + NumberAttribute, + UnicodeAttribute, + UnicodeSetAttribute, + UTCDateTimeAttribute, ) from datetime import datetime @@ -29,7 +35,7 @@ class Meta: answered = NumberAttribute(default=0) tags = UnicodeSetAttribute() last_post_datetime = UTCDateTimeAttribute(null=True) - notes = ListAttribute(default=list) + notes: ListAttribute[Any] = ListAttribute(default=list) # Delete the table @@ -60,7 +66,7 @@ class Meta: threads = [] for x in range(100): thread = Thread('forum-{0}'.format(x), 'subject-{0}'.format(x)) - thread.tags = ['tag1', 'tag2'] + thread.tags = {'tag1', 'tag2'} thread.last_post_datetime = datetime.now() threads.append(thread) @@ -75,16 +81,16 @@ class Meta: # Batch get item_keys = [('forum-{0}'.format(x), 'subject-{0}'.format(x)) for x in range(100)] -for item in Thread.batch_get(item_keys): - print(item) +for thread_item in Thread.batch_get(item_keys): + print(thread_item) # Scan -for item in Thread.scan(): - print(item) +for thread_item in Thread.scan(): + print(thread_item) # Query -for item in Thread.query('forum-1', Thread.subject.startswith('subject')): - print(item) +for thread_item in Thread.query('forum-1', Thread.subject.startswith('subject')): + print(thread_item) print("-"*80) @@ -103,11 +109,12 @@ class Meta: tags = UnicodeSetAttribute(attr_name='t') last_post_datetime = UTCDateTimeAttribute(attr_name='lp') + if not AliasedModel.exists(): AliasedModel.create_table(read_capacity_units=1, write_capacity_units=1, wait=True) # Create a thread -thread_item = AliasedModel( +aliased_thread_item = AliasedModel( 'Some Forum', 'Some Subject', tags=['foo', 'bar'], @@ -115,45 +122,45 @@ class Meta: ) # Save the thread -thread_item.save() +aliased_thread_item.save() # Batch write operation with AliasedModel.batch_write() as batch: - threads = [] - for x in range(100): - thread = AliasedModel('forum-{0}'.format(x), 'subject-{0}'.format(x)) - thread.tags = ['tag1', 'tag2'] - thread.last_post_datetime = datetime.now() - threads.append(thread) + aliased_threads = [] + for idx in range(100): + aliased_thread_item = AliasedModel('forum-{0}'.format(idx), 'subject-{0}'.format(idx)) + aliased_thread_item.tags = {'tag1', 'tag2'} + aliased_thread_item.last_post_datetime = datetime.now() + aliased_threads.append(aliased_thread_item) - for thread in threads: - batch.save(thread) + for aliased_thread_item in aliased_threads: + batch.save(aliased_thread_item) # Batch get item_keys = [('forum-{0}'.format(x), 'subject-{0}'.format(x)) for x in range(100)] -for item in AliasedModel.batch_get(item_keys): - print("Batch get item: {0}".format(item)) +for aliased_thread_item in AliasedModel.batch_get(item_keys): + print("Batch get item: {0}".format(aliased_thread_item)) # Scan -for item in AliasedModel.scan(): - print("Scanned item: {0}".format(item)) +for aliased_thread_item in AliasedModel.scan(): + print("Scanned item: {0}".format(aliased_thread_item)) # Query -for item in AliasedModel.query('forum-1', AliasedModel.subject.startswith('subject')): - print("Query using aliased attribute: {0}".format(item)) +for aliased_thread_item in AliasedModel.query('forum-1', AliasedModel.subject.startswith('subject')): + print("Query using aliased attribute: {0}".format(aliased_thread_item)) # Query with filters -for item in Thread.query('forum-1', (Thread.views == 0) | (Thread.replies == 0)): - print("Query result: {0}".format(item)) +for thread_item in Thread.query('forum-1', (Thread.views == 0) | (Thread.replies == 0)): + print("Query result: {0}".format(thread_item)) # Scan with filters -for item in Thread.scan(Thread.subject.startswith('subject') & (Thread.views == 0)): - print("Scanned item: {0} {1}".format(item.subject, item.views)) +for thread_item in Thread.scan(Thread.subject.startswith('subject') & (Thread.views == 0)): + print("Scanned item: {0} {1}".format(thread_item.subject, thread_item.views)) # Scan with null filter -for item in Thread.scan(Thread.subject.startswith('subject') & Thread.last_post_datetime.does_not_exist()): - print("Scanned item: {0} {1}".format(item.subject, item.views)) +for thread_item in Thread.scan(Thread.subject.startswith('subject') & Thread.last_post_datetime.does_not_exist()): + print("Scanned item: {0} {1}".format(thread_item.subject, thread_item.views)) # Conditionally save an item thread_item = Thread( diff --git a/examples/optimistic_locking.py b/examples/optimistic_locking.py index d80e9833a..0d0d0c0b8 100644 --- a/examples/optimistic_locking.py +++ b/examples/optimistic_locking.py @@ -46,7 +46,7 @@ def assert_condition_check_fails(): except TransactWriteError as e: assert isinstance(e.cause, ClientError) assert e.cause_response_code == "TransactionCanceledException" - assert "ConditionalCheckFailed" in e.cause_response_message + assert "ConditionalCheckFailed" in (e.cause_response_message or '') else: raise AssertionError("The version attribute conditional check should have failed.") diff --git a/mypy.ini b/mypy.ini index 5dfc38c9a..39bd93b50 100644 --- a/mypy.ini +++ b/mypy.ini @@ -10,6 +10,10 @@ warn_incomplete_stub = True follow_imports = normal show_error_codes = True +# Ignore errors in the docs/conf.py file +[mypy-conf] +ignore_errors = True + # TODO: burn these down [mypy-tests.*] ignore_errors = True diff --git a/pynamodb/indexes.py b/pynamodb/indexes.py index d923510d9..47ba6df6f 100644 --- a/pynamodb/indexes.py +++ b/pynamodb/indexes.py @@ -30,7 +30,7 @@ class Index(Generic[_M]): @classmethod def __init_subclass__(cls, **kwargs): - super().__init_subclass__(**kwargs) # type: ignore # see https://github.com/python/mypy/issues/4660 + super().__init_subclass__(**kwargs) if cls.Meta is not None: cls.Meta.attributes = {} for name, attribute in getmembers(cls, lambda o: isinstance(o, Attribute)): diff --git a/pynamodb/settings.py b/pynamodb/settings.py index 4a99885ba..49395a087 100644 --- a/pynamodb/settings.py +++ b/pynamodb/settings.py @@ -24,7 +24,7 @@ def _load_module(name, path): # https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly spec = importlib.util.spec_from_file_location(name, path) - module = importlib.util.module_from_spec(spec) + module = importlib.util.module_from_spec(spec) # type: ignore spec.loader.exec_module(module) # type: ignore return module diff --git a/pynamodb/transactions.py b/pynamodb/transactions.py index 950e89368..acc05fc9a 100644 --- a/pynamodb/transactions.py +++ b/pynamodb/transactions.py @@ -31,11 +31,11 @@ def __exit__(self, exc_type, exc_val, exc_tb): self._commit() -class TransactGet(Generic[_M], Transaction): +class TransactGet(Transaction): _results: Optional[List] = None - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: self._get_items: List[Dict] = [] self._futures: List[_ModelFuture] = [] super(TransactGet, self).__init__(*args, **kwargs) diff --git a/requirements-dev.txt b/requirements-dev.txt index 4b46fdeaa..f422d6348 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -4,8 +4,9 @@ pytest-mock # only used in CI coveralls -mypy==0.770;python_version>="3.7" +mypy==0.950;python_version>="3.7" pytest-cov # used for type-checking botocore-stubs +types-Flask diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index 5ba76383e..000000000 --- a/tests/conftest.py +++ /dev/null @@ -1,10 +0,0 @@ -import pytest - - -@pytest.fixture -def assert_mypy_output(pytestconfig): - pytest.importorskip('mypy') # we only install mypy in python>=3.6 tests - pytest.register_assert_rewrite('tests.mypy_helpers') - - from tests.mypy_helpers import assert_mypy_output - return lambda program: assert_mypy_output(program, use_pdb=pytestconfig.getoption('usepdb')) diff --git a/tests/test_mypy.py b/tests/test_mypy.py deleted file mode 100644 index 6d5851513..000000000 --- a/tests/test_mypy.py +++ /dev/null @@ -1,251 +0,0 @@ -""" -Note: The expected error strings may change in a future version of mypy. - Please update as needed. -""" - - -def test_model(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.models import Model - from pynamodb.expressions.operand import Path - - class MyModel(Model): - pass - - reveal_type(MyModel.count('hash', Path('a').between(1, 3))) # N: Revealed type is 'builtins.int' - """) - - -def test_model_query(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import NumberAttribute - from pynamodb.models import Model - - class MyModel(Model): - my_attr = NumberAttribute() - - # test conditions - MyModel.query(123, range_key_condition=(MyModel.my_attr == 5), filter_condition=(MyModel.my_attr == 5)) - - # test conditions are optional - MyModel.query(123, range_key_condition=None, filter_condition=None) - """) - - -def test_pagination(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import NumberAttribute - from pynamodb.models import Model - - class MyModel(Model): - my_attr = NumberAttribute() - - result_iterator = MyModel.query(123) - for model in result_iterator: - reveal_type(model) # N: Revealed type is '__main__.MyModel*' - if result_iterator.last_evaluated_key: - reveal_type(result_iterator.last_evaluated_key['my_attr']) # N: Revealed type is 'builtins.dict*[builtins.str, Any]' - """) - - -def test_model_update(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import NumberAttribute - from pynamodb.models import Model - - class MyModel(Model): - my_attr = NumberAttribute() - - my_model = MyModel() - my_model.update(actions=[ - # test update expressions - MyModel.my_attr.set(MyModel.my_attr + 123), - MyModel.my_attr.set(123 + MyModel.my_attr), - MyModel.my_attr.set(MyModel.my_attr - 123), - MyModel.my_attr.set(123 - MyModel.my_attr), - MyModel.my_attr.set(MyModel.my_attr | 123), - ]) - """) # noqa: E501 - - -def test_number_attribute(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import NumberAttribute - from pynamodb.models import Model - - class MyModel(Model): - my_attr = NumberAttribute() - - reveal_type(MyModel.my_attr) # N: Revealed type is 'pynamodb.attributes.NumberAttribute' - reveal_type(MyModel().my_attr) # N: Revealed type is 'builtins.float*' - """) - - -def test_unicode_attribute(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import UnicodeAttribute - from pynamodb.models import Model - - class MyModel(Model): - my_attr = UnicodeAttribute() - - reveal_type(MyModel.my_attr) # N: Revealed type is 'pynamodb.attributes.UnicodeAttribute' - reveal_type(MyModel().my_attr) # N: Revealed type is 'builtins.str*' - """) - - -def test_map_attribute(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import MapAttribute, UnicodeAttribute - from pynamodb.models import Model - - class MySubMap(MapAttribute): - s = UnicodeAttribute() - - class MyMap(MapAttribute): - m2 = MySubMap() - - class MyModel(Model): - m1 = MyMap() - - reveal_type(MyModel.m1) # N: Revealed type is '__main__.MyMap' - reveal_type(MyModel().m1) # N: Revealed type is '__main__.MyMap' - reveal_type(MyModel.m1.m2) # N: Revealed type is '__main__.MySubMap' - reveal_type(MyModel().m1.m2) # N: Revealed type is '__main__.MySubMap' - reveal_type(MyModel.m1.m2.s) # N: Revealed type is 'builtins.str*' - reveal_type(MyModel().m1.m2.s) # N: Revealed type is 'builtins.str*' - - reveal_type(MyMap.m2) # N: Revealed type is '__main__.MySubMap' - reveal_type(MyMap().m2) # N: Revealed type is '__main__.MySubMap' - - reveal_type(MySubMap.s) # N: Revealed type is 'pynamodb.attributes.UnicodeAttribute' - reveal_type(MySubMap().s) # N: Revealed type is 'builtins.str*' - """) - - -def test_list_attribute(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import ListAttribute, MapAttribute, UnicodeAttribute - from pynamodb.models import Model - - class MyMap(MapAttribute): - my_sub_attr = UnicodeAttribute() - - class MyModel(Model): - my_list = ListAttribute(of=MyMap) - my_untyped_list = ListAttribute() # E: Need type annotation for 'my_untyped_list' [var-annotated] - - reveal_type(MyModel.my_list) # N: Revealed type is 'pynamodb.attributes.ListAttribute[__main__.MyMap]' - reveal_type(MyModel().my_list) # N: Revealed type is 'builtins.list*[__main__.MyMap*]' - reveal_type(MyModel().my_list[0].my_sub_attr) # N: Revealed type is 'builtins.str*' - - # Untyped lists are not well supported yet - reveal_type(MyModel().my_untyped_list[0].my_sub_attr) # N: Revealed type is 'Any' - """) - - -def test_paths(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import ListAttribute, MapAttribute, UnicodeAttribute - from pynamodb.models import Model - - class MyMap(MapAttribute): - my_sub_attr = UnicodeAttribute() - - class MyModel(Model): - my_list = ListAttribute(of=MyMap) - my_map = MyMap() - - reveal_type(MyModel.my_list[0]) # N: Revealed type is 'pynamodb.expressions.operand.Path' - reveal_type(MyModel.my_list[0] == MyModel()) # N: Revealed type is 'pynamodb.expressions.condition.Comparison' - # the following string indexing is not type checked - not by mypy nor in runtime - reveal_type(MyModel.my_list[0]['my_sub_attr'] == 'foobar') # N: Revealed type is 'pynamodb.expressions.condition.Comparison' - reveal_type(MyModel.my_map == 'foobar') # N: Revealed type is 'pynamodb.expressions.condition.Comparison' - """) - - -def test_index_query_scan(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import NumberAttribute - from pynamodb.models import Model - from pynamodb.indexes import GlobalSecondaryIndex - from pynamodb.pagination import ResultIterator - - class UntypedIndex(GlobalSecondaryIndex): - bar = NumberAttribute(hash_key=True) - - class TypedIndex(GlobalSecondaryIndex[MyModel]): - bar = NumberAttribute(hash_key=True) - - class MyModel(Model): - foo = NumberAttribute(hash_key=True) - bar = NumberAttribute() - - untyped_index = UntypedIndex() - typed_index = TypedIndex() - - # Ensure old code keeps working - untyped_query_result: ResultIterator = MyModel.untyped_index.query(123) - model: MyModel = next(untyped_query_result) - not_model: int = next(untyped_query_result) # this is legacy behavior so it's "fine" - - # Allow users to specify which model their indices return - typed_query_result: ResultIterator[MyModel] = MyModel.typed_index.query(123) - my_model = next(typed_query_result) - not_model = next(typed_query_result) # E: Incompatible types in assignment (expression has type "MyModel", variable has type "int") [assignment] - - # Ensure old code keeps working - untyped_scan_result = MyModel.untyped_index.scan() - model = next(untyped_scan_result) - not_model = next(untyped_scan_result) # this is legacy behavior so it's "fine" - - # Allow users to specify which model their indices return - typed_scan_result = MyModel.typed_index.scan() - model = next(typed_scan_result) - not_model = next(typed_scan_result) # E: Incompatible types in assignment (expression has type "MyModel", variable has type "int") [assignment] - """) - - -def test_map_attribute_derivation(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.attributes import MapAttribute - - class MyMap(MapAttribute, object): - pass - """) - - -def test_is_in(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.models import Model - from pynamodb.attributes import UnicodeAttribute - - class MyModel(Model): - attr = UnicodeAttribute() - - _ = MyModel.attr.is_in('foo', 'bar') - _ = MyModel.attr.is_in(123) # E: Argument 1 to "is_in" of "Attribute" has incompatible type "int"; expected "str" [arg-type] - _ = MyModel.attr.is_in(['foo', 'bar']) # E: Argument 1 to "is_in" of "Attribute" has incompatible type "List[str]"; expected "str" [arg-type] - """) - - -def test_append(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.models import Model - from pynamodb.attributes import ListAttribute, NumberAttribute - - class MyModel(Model): - attr = ListAttribute(of=NumberAttribute) - - MyModel.attr.append(42) # E: Argument 1 to "append" of "Attribute" has incompatible type "int"; expected "Iterable[Any]" [arg-type] - MyModel.attr.append([42]) - MyModel.attr.prepend(42) # E: Argument 1 to "prepend" of "Attribute" has incompatible type "int"; expected "Iterable[Any]" [arg-type] - MyModel.attr.prepend([42]) - """) - -def test_transactions(assert_mypy_output): - assert_mypy_output(""" - from pynamodb.transactions import TransactWrite - with TransactWrite() as tx: - reveal_type(tx) # N: Revealed type is 'pynamodb.transactions.TransactWrite*' - """) diff --git a/typing_tests/__init__.py b/typing_tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/typing_tests/attributes.py b/typing_tests/attributes.py new file mode 100644 index 000000000..4765585ca --- /dev/null +++ b/typing_tests/attributes.py @@ -0,0 +1,72 @@ +from __future__ import annotations +from typing import Any + +from typing_extensions import assert_type + + +def test_number_attribute() -> None: + from pynamodb.attributes import NumberAttribute + from pynamodb.models import Model + + class MyModel(Model): + my_attr = NumberAttribute() + + assert_type(MyModel.my_attr, NumberAttribute) + assert_type(MyModel().my_attr, float) + + +def test_unicode_attribute() -> None: + from pynamodb.attributes import UnicodeAttribute + from pynamodb.models import Model + + class MyModel(Model): + my_attr = UnicodeAttribute() + + assert_type(MyModel.my_attr, UnicodeAttribute) + assert_type(MyModel().my_attr, str) + + +def test_map_attribute() -> None: + from pynamodb.attributes import MapAttribute, UnicodeAttribute + from pynamodb.models import Model + + class MySubMap(MapAttribute): + s = UnicodeAttribute() + + class MyMap(MapAttribute): + m2 = MySubMap() + + class MyModel(Model): + m1 = MyMap() + + assert_type(MyModel.m1, MyMap) + assert_type(MyModel().m1, MyMap) + assert_type(MyModel.m1.m2, MySubMap) + assert_type(MyModel().m1.m2, MySubMap) + assert_type(MyModel.m1.m2.s, str) + assert_type(MyModel().m1.m2.s, str) + + assert_type(MyMap.m2, MySubMap) + assert_type(MyMap().m2, MySubMap) + + assert_type(MySubMap.s, UnicodeAttribute) + assert_type(MySubMap().s, str) + + +def test_list_attribute() -> None: + from pynamodb.attributes import ListAttribute, MapAttribute, UnicodeAttribute + from pynamodb.models import Model + + class MyMap(MapAttribute): + my_sub_attr = UnicodeAttribute() + + class MyModel(Model): + my_list = ListAttribute(of=MyMap) + my_untyped_list = ListAttribute() # type: ignore[var-annotated] + + assert_type(MyModel.my_list, ListAttribute[MyMap]) + assert_type(MyModel().my_list, list[MyMap]) + assert_type(MyModel().my_list[0].my_sub_attr, str) + + # Untyped lists are not well-supported yet + assert_type(MyModel().my_untyped_list[0].my_sub_attr, Any) diff --git a/typing_tests/models.py b/typing_tests/models.py new file mode 100644 index 000000000..3ad4db3f0 --- /dev/null +++ b/typing_tests/models.py @@ -0,0 +1,149 @@ +from __future__ import annotations +from typing import Any + +from typing_extensions import assert_type + + +def test_model_count() -> None: + from pynamodb.models import Model + from pynamodb.expressions.operand import Path + + class MyModel(Model): + pass + + assert_type(MyModel.count('hash', Path('a').between(1, 3)), int) + + +def test_model_query() -> None: + from pynamodb.attributes import NumberAttribute + from pynamodb.models import Model + + class MyModel(Model): + my_attr = NumberAttribute() + + # test conditions + MyModel.query(123, range_key_condition=(MyModel.my_attr == 5), filter_condition=(MyModel.my_attr == 5)) + + # test conditions are optional + MyModel.query(123, range_key_condition=None, filter_condition=None) + + +def test_pagination() -> None: + from pynamodb.attributes import NumberAttribute + from pynamodb.models import Model + + class MyModel(Model): + my_attr = NumberAttribute() + + result_iterator = MyModel.query(123) + for model in result_iterator: + assert_type(model, MyModel) + if result_iterator.last_evaluated_key: + assert_type(result_iterator.last_evaluated_key['my_attr'], dict[str, Any]) + + +def test_model_update() -> None: + from pynamodb.attributes import NumberAttribute, UnicodeAttribute + from pynamodb.models import Model + + class MyModel(Model): + my_attr = NumberAttribute() + my_str_attr = UnicodeAttribute() + + my_model = MyModel() + my_model.update(actions=[ + # test update expressions + MyModel.my_attr.set(MyModel.my_attr + 123), + MyModel.my_attr.set(123 + MyModel.my_attr), + MyModel.my_attr.set(MyModel.my_attr - 123), + MyModel.my_attr.set(123 - MyModel.my_attr), + MyModel.my_attr.set(MyModel.my_attr | 123), + ]) + + +def test_paths() -> None: + import pynamodb.expressions.operand + import pynamodb.expressions.condition + from pynamodb.attributes import ListAttribute, MapAttribute, UnicodeAttribute + from pynamodb.models import Model + + class MyMap(MapAttribute): + my_sub_attr = UnicodeAttribute() + + class MyModel(Model): + my_list = ListAttribute(of=MyMap) + my_map = MyMap() + + assert_type(MyModel.my_list[0], pynamodb.expressions.operand.Path) + assert_type(MyModel.my_list[0] == MyModel(), pynamodb.expressions.condition.Comparison) + # the following string indexing is not type checked - not by mypy nor in runtime + assert_type(MyModel.my_list[0]['my_sub_attr'] == 'foobar', pynamodb.expressions.condition.Comparison) + assert_type(MyModel.my_map == 'foobar', pynamodb.expressions.condition.Comparison) + + +def test_index_query_scan() -> None: + from pynamodb.attributes import NumberAttribute + from pynamodb.models import Model + from pynamodb.indexes import GlobalSecondaryIndex + from pynamodb.pagination import ResultIterator + + class UntypedIndex(GlobalSecondaryIndex): + bar = NumberAttribute(hash_key=True) + + class TypedIndex(GlobalSecondaryIndex['MyModel']): + bar = NumberAttribute(hash_key=True) + + class MyModel(Model): + foo = NumberAttribute(hash_key=True) + bar = NumberAttribute() + + untyped_index = UntypedIndex() + typed_index = TypedIndex() + + # Ensure old code keeps working + untyped_query_result: ResultIterator = MyModel.untyped_index.query(123) + assert_type(next(untyped_query_result), Any) + + # Allow users to specify which model their indices return + typed_query_result: ResultIterator[MyModel] = MyModel.typed_index.query(123) + assert_type(next(typed_query_result), MyModel) + + # Ensure old code keeps working + untyped_scan_result = MyModel.untyped_index.scan() + assert_type(next(untyped_scan_result), Any) + + # Allow users to specify which model their indices return + typed_scan_result = MyModel.typed_index.scan() + assert_type(next(typed_scan_result), MyModel) + + +def test_map_attribute_derivation() -> None: + from pynamodb.attributes import MapAttribute + + class MyMap(MapAttribute, object): + pass + + +def test_is_in() -> None: + from pynamodb.models import Model + from pynamodb.attributes import UnicodeAttribute + + class MyModel(Model): + attr = UnicodeAttribute() + + _ = MyModel.attr.is_in('foo', 'bar') + _ = MyModel.attr.is_in(123) # type:ignore[arg-type] + _ = MyModel.attr.is_in(['foo', 'bar']) # type:ignore[arg-type] + + +def test_append() -> None: + from pynamodb.models import Model + from pynamodb.attributes import ListAttribute, NumberAttribute + + class MyModel(Model): + attr = ListAttribute(of=NumberAttribute) + + MyModel.attr.append(42) # type:ignore[arg-type] + MyModel.attr.append([42]) + MyModel.attr.prepend(42) # type:ignore[arg-type] + MyModel.attr.prepend([42]) diff --git a/typing_tests/transactions.py b/typing_tests/transactions.py new file mode 100644 index 000000000..1b1bd3ad4 --- /dev/null +++ b/typing_tests/transactions.py @@ -0,0 +1,28 @@ +from typing_extensions import assert_type + + +def test_transact_write() -> None: + from pynamodb.transactions import TransactWrite + with TransactWrite() as tx: + assert_type(tx, TransactWrite) + + +def test_transact_get() -> None: + from pynamodb.transactions import TransactGet + from pynamodb.models import Model, _ModelFuture + + class FirstModel(Model): + pass + + class SecondModel(Model): + pass + + with TransactGet() as tx: + assert_type(tx, TransactGet) + assert_type(tx.get(FirstModel, "pk"), _ModelFuture[FirstModel]) + assert_type(tx.get(SecondModel, "pk"), _ModelFuture[SecondModel]) + + second_model_instance_future = tx.get(SecondModel, "pk") + + assert_type(second_model_instance_future.get(), SecondModel) + _first_model_instance: FirstModel = second_model_instance_future.get() # type:ignore[assignment]