Add parameters to typing.List.

This commit is contained in:
Andrey Rakhmatullin 2024-05-31 21:20:22 +05:00
parent da42e8f124
commit 98c755e5fb
4 changed files with 19 additions and 13 deletions

View File

@ -4,7 +4,7 @@ import json
import logging
from abc import abstractmethod
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, Type, cast
from typing import TYPE_CHECKING, Any, List, Optional, Type, cast
from twisted.internet.defer import Deferred
@ -362,13 +362,13 @@ class Scheduler(BaseScheduler):
return str(dqdir)
return None
def _read_dqs_state(self, dqdir: str) -> list:
def _read_dqs_state(self, dqdir: str) -> List[int]:
path = Path(dqdir, "active.json")
if not path.exists():
return []
with path.open(encoding="utf-8") as f:
return cast(list, json.load(f))
return cast(List[int], json.load(f))
def _write_dqs_state(self, dqdir: str, state: list) -> None:
def _write_dqs_state(self, dqdir: str, state: List[int]) -> None:
with Path(dqdir, "active.json").open("w", encoding="utf-8") as f:
json.dump(state, f)

View File

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Dict, Union
from typing import TYPE_CHECKING, Dict, List, Tuple, Union
from twisted.web import http
@ -17,7 +17,9 @@ if TYPE_CHECKING:
from typing_extensions import Self
def get_header_size(headers: Dict[str, Union[list, tuple]]) -> int:
def get_header_size(
headers: Dict[str, Union[List[Union[str, bytes]], Tuple[Union[str, bytes], ...]]]
) -> int:
size = 0
for key, value in headers.items():
if isinstance(value, (list, tuple)):

View File

@ -1,14 +1,18 @@
from typing import AsyncGenerator, AsyncIterable, Iterable, Union
from typing import AsyncGenerator, AsyncIterable, Iterable, List, TypeVar, Union
_T = TypeVar("_T")
async def collect_asyncgen(result: AsyncIterable) -> list:
async def collect_asyncgen(result: AsyncIterable[_T]) -> List[_T]:
results = []
async for x in result:
results.append(x)
return results
async def as_async_generator(it: Union[Iterable, AsyncIterable]) -> AsyncGenerator:
async def as_async_generator(
it: Union[Iterable[_T], AsyncIterable[_T]]
) -> AsyncGenerator[_T, None]:
"""Wraps an iterable (sync or async) into an async generator."""
if isinstance(it, AsyncIterable):
async for r in it:

View File

@ -46,7 +46,7 @@ _KT = TypeVar("_KT")
_VT = TypeVar("_VT")
def flatten(x: Iterable) -> list:
def flatten(x: Iterable[Any]) -> List[Any]:
"""flatten(sequence) -> list
Returns a single, flat list which contains all elements retrieved
@ -66,7 +66,7 @@ def flatten(x: Iterable) -> list:
return list(iflatten(x))
def iflatten(x: Iterable) -> Iterable:
def iflatten(x: Iterable[Any]) -> Iterable[Any]:
"""iflatten(sequence) -> iterator
Similar to ``.flatten()``, but returns iterator instead"""
@ -101,10 +101,10 @@ def is_listlike(x: Any) -> bool:
return hasattr(x, "__iter__") and not isinstance(x, (str, bytes))
def unique(list_: Iterable, key: Callable[[Any], Any] = lambda x: x) -> list:
def unique(list_: Iterable[_T], key: Callable[[_T], Any] = lambda x: x) -> List[_T]:
"""efficient function to uniquify a list preserving item order"""
seen = set()
result = []
result: List[_T] = []
for item in list_:
seenkey = key(item)
if seenkey in seen: