mirror of https://github.com/celery/kombu.git
Annotate `utils/uuid.py` and `utils/text.py` (#1514)
* Annotate `utils/uuid.py` and `utils/text.py` * Use callable without any parameter * Update kombu/utils/text.py
This commit is contained in:
parent
894ddfc8b6
commit
7d99b6f79f
|
@ -3,6 +3,7 @@
|
|||
|
||||
|
||||
from difflib import SequenceMatcher
|
||||
from typing import Iterator, Optional, Sequence, Tuple, Union
|
||||
|
||||
from kombu import version_info_t
|
||||
|
||||
|
@ -16,8 +17,7 @@ def escape_regex(p, white=''):
|
|||
for c in p)
|
||||
|
||||
|
||||
def fmatch_iter(needle, haystack, min_ratio=0.6):
|
||||
# type: (str, Sequence[str], float) -> Iterator[Tuple[float, str]]
|
||||
def fmatch_iter(needle: str, haystack: Sequence[str], min_ratio: float = 0.6) -> Iterator[Tuple[float, str]]:
|
||||
"""Fuzzy match: iteratively.
|
||||
|
||||
Yields:
|
||||
|
@ -29,19 +29,17 @@ def fmatch_iter(needle, haystack, min_ratio=0.6):
|
|||
yield ratio, key
|
||||
|
||||
|
||||
def fmatch_best(needle, haystack, min_ratio=0.6):
|
||||
# type: (str, Sequence[str], float) -> str
|
||||
def fmatch_best(needle: str, haystack: Sequence[str], min_ratio: float = 0.6) -> Optional[str]:
|
||||
"""Fuzzy match - Find best match (scalar)."""
|
||||
try:
|
||||
return sorted(
|
||||
fmatch_iter(needle, haystack, min_ratio), reverse=True,
|
||||
)[0][1]
|
||||
except IndexError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def version_string_as_tuple(s):
|
||||
# type: (str) -> version_info_t
|
||||
def version_string_as_tuple(s: str) -> version_info_t:
|
||||
"""Convert version string to version info tuple."""
|
||||
v = _unpack_version(*s.split('.'))
|
||||
# X.Y.3a1 -> (X, Y, 3, 'a1')
|
||||
|
@ -53,13 +51,17 @@ def version_string_as_tuple(s):
|
|||
return v
|
||||
|
||||
|
||||
def _unpack_version(major, minor=0, micro=0, releaselevel='', serial=''):
|
||||
# type: (int, int, int, str, str) -> version_info_t
|
||||
def _unpack_version(
|
||||
major: str,
|
||||
minor: Union[str, int] = 0,
|
||||
micro: Union[str, int] = 0,
|
||||
releaselevel: str = '',
|
||||
serial: str = ''
|
||||
) -> version_info_t:
|
||||
return version_info_t(int(major), int(minor), micro, releaselevel, serial)
|
||||
|
||||
|
||||
def _splitmicro(micro, releaselevel='', serial=''):
|
||||
# type: (int, str, str) -> Tuple[int, str, str]
|
||||
def _splitmicro(micro: str, releaselevel: str = '', serial: str = '') -> Tuple[int, str, str]:
|
||||
for index, char in enumerate(micro):
|
||||
if not char.isdigit():
|
||||
break
|
||||
|
|
|
@ -1,9 +1,9 @@
|
|||
"""UUID utilities."""
|
||||
|
||||
from uuid import uuid4
|
||||
from typing import Callable
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
|
||||
def uuid(_uuid=uuid4):
|
||||
def uuid(_uuid: Callable[[], UUID] = uuid4) -> str:
|
||||
"""Generate unique id in UUID4 format.
|
||||
|
||||
See Also:
|
||||
|
|
|
@ -22,8 +22,12 @@ disallow_untyped_defs = True
|
|||
ignore_missing_imports = True
|
||||
files =
|
||||
kombu/utils/time.py,
|
||||
kombu/utils/uuid.py,
|
||||
t/unit/utils/test_uuid.py,
|
||||
kombu/utils/text.py,
|
||||
kombu/asynchronous/semaphore.py
|
||||
|
||||
|
||||
[pep257]
|
||||
ignore = D102,D104,D203,D105,D213
|
||||
|
||||
|
|
|
@ -3,10 +3,10 @@ from kombu.utils.uuid import uuid
|
|||
|
||||
class test_UUID:
|
||||
|
||||
def test_uuid4(self):
|
||||
def test_uuid4(self) -> None:
|
||||
assert uuid() != uuid()
|
||||
|
||||
def test_uuid(self):
|
||||
def test_uuid(self) -> None:
|
||||
i1 = uuid()
|
||||
i2 = uuid()
|
||||
assert isinstance(i1, str)
|
||||
|
|
Loading…
Reference in New Issue