Merge pull request #37 from sbidoul/mypy-0930-sbi

Benefit from mypy 0.930
This commit is contained in:
Stéphane Bidoul 2021-12-22 19:55:41 +01:00 committed by GitHub
commit 4dba094949
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 9 additions and 10 deletions

View file

@ -30,7 +30,7 @@ test = [
"pytest-mock",
]
mypy = [
"mypy",
"mypy>=0.930",
]
[project.urls]

View file

@ -1,4 +1,4 @@
# frozen requirements generated by pip-deepfreeze
mypy==0.910
mypy==0.930
mypy-extensions==0.4.3
toml==0.10.2
tomli==1.2.2

View file

@ -17,6 +17,7 @@ from kubernetes.client.exceptions import ApiException
from kubernetes.client.models.v1_deployment import V1Deployment
from kubernetes.client.models.v1_job import V1Job
from pydantic import BaseModel
from typing_extensions import NotRequired
from .github import CommitInfo
from .settings import BuildSettings, settings
@ -56,10 +57,10 @@ def delete_deployment(deployment_name: str) -> None:
)
class PatchOperation(TypedDict, total=False):
class PatchOperation(TypedDict):
op: str
path: str
value: str | int # maybe absent, hence total=False above
value: NotRequired[str | int]
@sync_to_async

View file

@ -21,14 +21,12 @@ def slugify(s: str | int) -> str:
return re.sub(r"[^a-z0-9]", "-", str(s).lower())
# TODO replace ... with P below when mypy supports PEP 612
# (https://github.com/python/mypy/issues/8645)
P = ParamSpec("P")
R = TypeVar("R")
T = TypeVar("T")
def sync_to_async(func: Callable[..., R]) -> Callable[..., Awaitable[R]]:
def sync_to_async(func: Callable[P, R]) -> Callable[P, Awaitable[R]]:
@wraps(func)
async def inner(*args: Any, **kwargs: Any) -> R:
f = functools.partial(func, *args, **kwargs)
@ -38,8 +36,8 @@ def sync_to_async(func: Callable[..., R]) -> Callable[..., Awaitable[R]]:
def sync_to_async_iterator(
iterator_func: Callable[..., Generator[R, None, None]]
) -> Callable[..., AsyncGenerator[R, None]]:
iterator_func: Callable[P, Generator[R, None, None]]
) -> Callable[P, AsyncGenerator[R, None]]:
@sync_to_async
def async_next(iterator: Iterator[R]) -> R:
try: