complete draft: update from Atlas

15-more-types/cafeteria/contravariant.py

@@ -38,8 +38,7 @@ deploy(trash_can)
 # tag::DEPLOY_NOT_VALID[]
 compost_can: TrashCan[Compostable] = TrashCan()
 deploy(compost_can)
-# end::DEPLOY_NOT_VALID[]
-
-## Argument 1 to "deploy" has
+## mypy: Argument 1 to "deploy" has
 ## incompatible type "TrashCan[Compostable]"
 ##          expected "TrashCan[Biodegradable]"
+# end::DEPLOY_NOT_VALID[]

15-more-types/cafeteria/covariant.py

@@ -38,11 +38,12 @@ orange_juice_dispenser = BeverageDispenser(OrangeJuice())
 install(orange_juice_dispenser)
 # end::INSTALL_JUICE_DISPENSERS[]
 
-################################################ not a juice dispenser
+################################################ more general dispenser
 
+# tag::INSTALL_BEVERAGE_DISPENSER[]
 beverage_dispenser = BeverageDispenser(Beverage())
-
-## Argument 1 to "install" has
+install(beverage_dispenser)
+## mypy: Argument 1 to "install" has
 ## incompatible type "BeverageDispenser[Beverage]"
 ##          expected "BeverageDispenser[Juice]"
-install(beverage_dispenser)
+# end::INSTALL_BEVERAGE_DISPENSER[]

15-more-types/cafeteria/invariant.py

@@ -37,7 +37,7 @@ install(juice_dispenser)
 # tag::INSTALL_BEVERAGE_DISPENSER[]
 beverage_dispenser = BeverageDispenser(Beverage())
 install(beverage_dispenser)
-## Argument 1 to "install" has
+## mypy: Argument 1 to "install" has
 ## incompatible type "BeverageDispenser[Beverage]"
 ##          expected "BeverageDispenser[Juice]"
 # end::INSTALL_BEVERAGE_DISPENSER[]
@@ -48,7 +48,7 @@ install(beverage_dispenser)
 # tag::INSTALL_ORANGE_JUICE_DISPENSER[]
 orange_juice_dispenser = BeverageDispenser(OrangeJuice())
 install(orange_juice_dispenser)
-# end::INSTALL_ORANGE_JUICE_DISPENSER[]
-## Argument 1 to "install" has
+## mypy: Argument 1 to "install" has
 ## incompatible type "BeverageDispenser[OrangeJuice]"
 ##          expected "BeverageDispenser[Juice]"
+# end::INSTALL_ORANGE_JUICE_DISPENSER[]

15-more-types/cast/empty.py

@@ -0,0 +1,2 @@
+# Mypy 0.812 can't spot this inevitable runtime IndexError
+print([][0])

15-more-types/cast/find.py

@@ -0,0 +1,23 @@
+# tag::CAST[]
+from typing import cast
+
+def find_first_str(a: list[object]) -> str:
+    index = next(i for i, x in enumerate(a) if isinstance(x, str))
+    # We only get here if there's at least one string in a
+    return cast(str, a[index])
+# end::CAST[]
+
+
+from typing import TYPE_CHECKING
+
+l1 = [10, 20, 'thirty', 40]
+if TYPE_CHECKING:
+    reveal_type(l1)
+
+print(find_first_str(l1))
+
+l2 = [0, ()]
+try:
+    find_first_str(l2)
+except StopIteration as e:
+    print(repr(e))

15-more-types/cast/tcp_echo.py

@@ -0,0 +1,37 @@
+import asyncio
+
+from asyncio import StreamReader, StreamWriter
+
+# tag::CAST_IMPORTS[]
+from asyncio.trsock import TransportSocket
+from typing import cast
+# end::CAST_IMPORTS[]
+
+async def handle_echo(reader: StreamReader, writer: StreamWriter) -> None:
+    data = await reader.read(100)
+    message = data.decode()
+    addr = writer.get_extra_info('peername')
+
+    print(f"Received {message!r} from {addr!r}")
+
+    print(f"Send: {message!r}")
+    writer.write(data)
+    await writer.drain()
+
+    print("Close the connection")
+    writer.close()
+
+async def main() -> None:
+    server = await asyncio.start_server(
+        handle_echo, '127.0.0.1', 8888)
+
+    # tag::CAST_USE[]
+    socket_list = cast(tuple[TransportSocket, ...], server.sockets)
+    addr = socket_list[0].getsockname()
+    # end::CAST_USE[]
+    print(f'Serving on {addr}')
+
+    async with server:
+        await server.serve_forever()
+
+asyncio.run(main())

15-more-types/cast/tcp_echo_no_cast.py

@@ -0,0 +1,29 @@
+import asyncio
+
+from asyncio import StreamReader, StreamWriter
+
+async def handle_echo(reader: StreamReader, writer: StreamWriter) -> None:
+    data = await reader.read(100)
+    message = data.decode()
+    addr = writer.get_extra_info('peername')
+
+    print(f"Received {message!r} from {addr!r}")
+
+    print(f"Send: {message!r}")
+    writer.write(data)
+    await writer.drain()
+
+    print("Close the connection")
+    writer.close()
+
+async def main() -> None:
+    server = await asyncio.start_server(
+        handle_echo, '127.0.0.1', 8888)
+
+    addr = server.sockets[0].getsockname()
+    print(f'Serving on {addr}')
+
+    async with server:
+        await server.serve_forever()
+
+asyncio.run(main())

15-more-types/clip_annot_demo.py

@@ -0,0 +1,3 @@
+from clip_annot_post import clip
+
+print(clip.__annotations__)

15-more-types/clip_annot_post.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 """
     >>> clip('banana ', 6)
     'banana'
@@ -18,9 +20,9 @@
 """
 
 # tag::CLIP_ANNOT[]
-
-def clip(text: str, max_len: 'int > 0' = 80) -> str:  # <1>
-    """Return text clipped at the last space before or after max_len
+def clip(text: str, max_len: int = 80) -> str:
+    """Return new ``str`` clipped at last space before or after ``max_len``.
+       Return full ``text`` if no space found.
     """
     end = None
     if len(text) > max_len:
@@ -31,7 +33,7 @@ def clip(text: str, max_len: 'int > 0' = 80) -> str:  # <1>
             space_after = text.rfind(' ', max_len)
             if space_after >= 0:
                 end = space_after
-    if end is None:  # no spaces were found
+    if end is None:
         end = len(text)
     return text[:end].rstrip()
 

15-more-types/clip_annot_signature.rst

@@ -1,10 +0,0 @@
->>> from clip_annot import clip
->>> from inspect import signature
->>> sig = signature(clip)
->>> sig.return_annotation
-<class 'str'>
->>> for param in sig.parameters.values():
-...     note = repr(param.annotation).ljust(13)
-...     print(note, ':', param.name, '=', param.default)
-<class 'str'> : text = <class 'inspect._empty'>
-'int > 0'     : max_len = 80

15-more-types/lotto/generic_lotto.py

@@ -0,0 +1,29 @@
+import random
+
+from collections.abc import Iterable
+from typing import TypeVar, Generic
+
+from tombola import Tombola
+
+T = TypeVar('T')
+
+class LottoBlower(Tombola, Generic[T]):  # <1>
+
+    def __init__(self, items: Iterable[T]) -> None:  # <2>
+        self._balls = list[T](items)
+
+    def load(self, items: Iterable[T]) -> None:  # <3>
+        self._balls.extend(items)
+
+    def pick(self) -> T:  # <4>
+        try:
+            position = random.randrange(len(self._balls))
+        except ValueError:
+            raise LookupError('pick from empty LottoBlower')
+        return self._balls.pop(position)
+
+    def loaded(self) -> bool:  # <5>
+        return bool(self._balls)
+
+    def inspect(self) -> tuple[T, ...]:  # <6>
+        return tuple(self._balls)

15-more-types/lotto/generic_lotto_demo.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+
+from typing import TYPE_CHECKING
+
+# tag::LOTTO_USE[]
+from generic_lotto import LottoBlower
+
+machine = LottoBlower[int](range(1, 11))  # <1>
+
+first = machine.pick()  # <2>
+remain = machine.inspect()  # <3>
+# end::LOTTO_USE[]
+
+expected = set(i for i in range(1, 11) if i != first)
+
+assert set(remain) == expected
+
+print('picked:', first)
+print('remain:', remain)
+
+if TYPE_CHECKING:
+    reveal_type(first)
+    # Revealed type is 'builtins.int*'
+if TYPE_CHECKING:
+    reveal_type(remain)
+    # Revealed type is 'builtins.tuple[builtins.int*]'
+
+

15-more-types/lotto/generic_lotto_errors.py

@@ -0,0 +1,18 @@
+from generic_lotto import LottoBlower
+
+machine = LottoBlower[int]([1, .2])
+## error: List item 1 has incompatible type "float";  # <1>
+##        expected "int"
+
+machine = LottoBlower[int](range(1, 11))
+
+machine.load('ABC')
+## error: Argument 1 to "load" of "LottoBlower"  # <2>
+##        has incompatible type "str";
+##        expected "Iterable[int]"
+## note:  Following member(s) of "str" have conflicts:
+## note:      Expected:
+## note:          def __iter__(self) -> Iterator[int]
+## note:      Got:
+## note:          def __iter__(self) -> Iterator[str]
+

15-more-types/lotto/tombola.py

@@ -0,0 +1,34 @@
+# tag::TOMBOLA_ABC[]
+
+import abc
+
+class Tombola(abc.ABC):  # <1>
+
+    @abc.abstractmethod
+    def load(self, iterable):  # <2>
+        """Add items from an iterable."""
+
+    @abc.abstractmethod
+    def pick(self):  # <3>
+        """Remove item at random, returning it.
+
+        This method should raise `LookupError` when the instance is empty.
+        """
+
+    def loaded(self):  # <4>
+        """Return `True` if there's at least 1 item, `False` otherwise."""
+        return bool(self.inspect())  # <5>
+
+    def inspect(self):
+        """Return a sorted tuple with the items currently inside."""
+        items = []
+        while True:  # <6>
+            try:
+                items.append(self.pick())
+            except LookupError:
+                break
+        self.load(items)  # <7>
+        return tuple(items)
+
+
+# end::TOMBOLA_ABC[]

15-more-types/mysum.py

@@ -1,13 +1,14 @@
-from functools import reduce  # <1>
-from operator import add
-from typing import overload, Iterable, Union, TypeVar
+import functools
+import operator
+from collections.abc import Iterable
+from typing import overload, Union, TypeVar
 
 T = TypeVar('T')
-S = TypeVar('S')  # <2>
+S = TypeVar('S')  # <1>
 
 @overload
-def sum(it: Iterable[T]) -> Union[T, int]: ...  # <3>
+def sum(it: Iterable[T]) -> Union[T, int]: ...  # <2>
 @overload
-def sum(it: Iterable[T], /, start: S) -> Union[T, S]: ...  # <4>
-def sum(it, /, start=0):  # <5>
-    return reduce(add, it, start)
+def sum(it: Iterable[T], /, start: S) -> Union[T, S]: ...  # <3>
+def sum(it, /, start=0):  # <4>
+    return functools.reduce(operator.add, it, start)

15-more-types/protocol/abs_demo.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+
+import math
+from typing import NamedTuple, SupportsAbs
+
+class Vector2d(NamedTuple):
+    x: float
+    y: float
+
+    def __abs__(self) -> float:  # <1>
+        return math.hypot(self.x, self.y)
+
+def is_unit(v: SupportsAbs[float]) -> bool:  # <2>
+    """'True' if the magnitude of 'v' is close to 1."""
+    return math.isclose(abs(v), 1.0)  # <3>
+
+assert issubclass(Vector2d, SupportsAbs)  # <4>
+
+v0 = Vector2d(0, 1)  # <5>
+sqrt2 = math.sqrt(2)
+v1 = Vector2d(sqrt2 / 2, sqrt2 / 2)
+v2 = Vector2d(1, 1)
+v3 = complex(.5, math.sqrt(3) / 2)
+v4 = 1  # <6>
+
+assert is_unit(v0)
+assert is_unit(v1)
+assert not is_unit(v2)
+assert is_unit(v3)
+assert is_unit(v4)
+
+print('OK')

15-more-types/protocol/mymax/mymax.py

@@ -12,10 +12,10 @@ MISSING = object()
 EMPTY_MSG = 'max() arg is an empty sequence'
 
 @overload
-def max(__arg1: LT, __arg2: LT, *_args: LT, key: None = ...) -> LT:
+def max(__arg1: LT, __arg2: LT, *args: LT, key: None = ...) -> LT:
     ...
 @overload
-def max(__arg1: T, __arg2: T, *_args: T, key: Callable[[T], LT]) -> T:
+def max(__arg1: T, __arg2: T, *args: T, key: Callable[[T], LT]) -> T:
     ...
 @overload
 def max(__iterable: Iterable[LT], *, key: None = ...) -> LT:

15-more-types/protocol/random/generic_randompick.py

@@ -0,0 +1,7 @@
+from typing import Protocol, runtime_checkable, TypeVar
+
+T_co = TypeVar('T_co', covariant=True)  # <1>
+
+@runtime_checkable
+class RandomPicker(Protocol[T_co]):  # <2>
+    def pick(self) -> T_co: ...  # <3>

15-more-types/protocol/random/generic_randompick_test.py

@@ -1,24 +1,26 @@
 import random
-from typing import Iterable, TYPE_CHECKING
+from typing import Iterable, Generic, TypeVar, TYPE_CHECKING
 
-from randompick_generic import GenericRandomPicker
+T_co = TypeVar('T_co', covariant=True)
+
+from generic_randompick import RandomPicker
 
 
-class LottoPicker:
-    def __init__(self, items: Iterable[int]) -> None:
+class LottoPicker(Generic[T_co]):
+    def __init__(self, items: Iterable[T_co]) -> None:
         self._items = list(items)
         random.shuffle(self._items)
 
-    def pick(self) -> int:
+    def pick(self) -> T_co:
         return self._items.pop()
 
 
 def test_issubclass() -> None:
-    assert issubclass(LottoPicker, GenericRandomPicker)
+    assert issubclass(LottoPicker, RandomPicker)
 
 
 def test_isinstance() -> None:
-    popper: GenericRandomPicker = LottoPicker([1])
+    popper: RandomPicker = LottoPicker[int]([1])
     if TYPE_CHECKING:
         reveal_type(popper)
         # Revealed type is '???'

15-more-types/randompick_generic.py

@@ -1,7 +0,0 @@
-from typing import Protocol, runtime_checkable, TypeVar
-
-T_co = TypeVar('T_co', covariant=True)
-
-@runtime_checkable
-class GenericRandomPicker(Protocol[T_co]):
-    def pick(self) -> T_co: ...