mirror of
https://github.com/ansible-collections/community.docker.git
synced 2026-07-29 11:55:04 +00:00
Add typing information, 1/2 (#1176)
* Re-enable typing and improve config. * Make mypy pass. * Improve settings. * First batch of types. * Add more type hints. * Fixes. * Format. * Fix split_port() without returning to previous type chaos. * Continue with type hints (and ignores).
This commit is contained in:
@@ -12,6 +12,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import typing as t
|
||||
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from collections.abc import Collection, Sequence
|
||||
|
||||
|
||||
PORT_SPEC = re.compile(
|
||||
@@ -26,32 +31,42 @@ PORT_SPEC = re.compile(
|
||||
)
|
||||
|
||||
|
||||
def add_port_mapping(port_bindings, internal_port, external):
|
||||
def add_port_mapping(
|
||||
port_bindings: dict[str, list[str | tuple[str, str | None] | None]],
|
||||
internal_port: str,
|
||||
external: str | tuple[str, str | None] | None,
|
||||
) -> None:
|
||||
if internal_port in port_bindings:
|
||||
port_bindings[internal_port].append(external)
|
||||
else:
|
||||
port_bindings[internal_port] = [external]
|
||||
|
||||
|
||||
def add_port(port_bindings, internal_port_range, external_range):
|
||||
def add_port(
|
||||
port_bindings: dict[str, list[str | tuple[str, str | None] | None]],
|
||||
internal_port_range: list[str],
|
||||
external_range: list[str] | list[tuple[str, str | None]] | None,
|
||||
) -> None:
|
||||
if external_range is None:
|
||||
for internal_port in internal_port_range:
|
||||
add_port_mapping(port_bindings, internal_port, None)
|
||||
else:
|
||||
ports = zip(internal_port_range, external_range)
|
||||
for internal_port, external_port in ports:
|
||||
add_port_mapping(port_bindings, internal_port, external_port)
|
||||
for internal_port, external_port in zip(internal_port_range, external_range):
|
||||
# mypy loses the exact type of eternal_port elements for some reason...
|
||||
add_port_mapping(port_bindings, internal_port, external_port) # type: ignore
|
||||
|
||||
|
||||
def build_port_bindings(ports):
|
||||
port_bindings = {}
|
||||
def build_port_bindings(
|
||||
ports: Collection[str],
|
||||
) -> dict[str, list[str | tuple[str, str | None] | None]]:
|
||||
port_bindings: dict[str, list[str | tuple[str, str | None] | None]] = {}
|
||||
for port in ports:
|
||||
internal_port_range, external_range = split_port(port)
|
||||
add_port(port_bindings, internal_port_range, external_range)
|
||||
return port_bindings
|
||||
|
||||
|
||||
def _raise_invalid_port(port):
|
||||
def _raise_invalid_port(port: str) -> t.NoReturn:
|
||||
raise ValueError(
|
||||
f'Invalid port "{port}", should be '
|
||||
"[[remote_ip:]remote_port[-remote_port]:]"
|
||||
@@ -59,39 +74,64 @@ def _raise_invalid_port(port):
|
||||
)
|
||||
|
||||
|
||||
def port_range(start, end, proto, randomly_available_port=False):
|
||||
if not start:
|
||||
@t.overload
|
||||
def port_range(
|
||||
start: str,
|
||||
end: str | None,
|
||||
proto: str,
|
||||
randomly_available_port: bool = False,
|
||||
) -> list[str]: ...
|
||||
|
||||
|
||||
@t.overload
|
||||
def port_range(
|
||||
start: str | None,
|
||||
end: str | None,
|
||||
proto: str,
|
||||
randomly_available_port: bool = False,
|
||||
) -> list[str] | None: ...
|
||||
|
||||
|
||||
def port_range(
|
||||
start: str | None,
|
||||
end: str | None,
|
||||
proto: str,
|
||||
randomly_available_port: bool = False,
|
||||
) -> list[str] | None:
|
||||
if start is None:
|
||||
return start
|
||||
if not end:
|
||||
if end is None:
|
||||
return [f"{start}{proto}"]
|
||||
if randomly_available_port:
|
||||
return [f"{start}-{end}{proto}"]
|
||||
return [f"{port}{proto}" for port in range(int(start), int(end) + 1)]
|
||||
|
||||
|
||||
def split_port(port):
|
||||
if hasattr(port, "legacy_repr"):
|
||||
# This is the worst hack, but it prevents a bug in Compose 1.14.0
|
||||
# https://github.com/docker/docker-py/issues/1668
|
||||
# TODO: remove once fixed in Compose stable
|
||||
port = port.legacy_repr()
|
||||
def split_port(
|
||||
port: str,
|
||||
) -> tuple[list[str], list[str] | list[tuple[str, str | None]] | None]:
|
||||
port = str(port)
|
||||
match = PORT_SPEC.match(port)
|
||||
if match is None:
|
||||
_raise_invalid_port(port)
|
||||
parts = match.groupdict()
|
||||
|
||||
host = parts["host"]
|
||||
proto = parts["proto"] or ""
|
||||
internal = port_range(parts["int"], parts["int_end"], proto)
|
||||
external = port_range(parts["ext"], parts["ext_end"], "", len(internal) == 1)
|
||||
host: str | None = parts["host"]
|
||||
proto: str = parts["proto"] or ""
|
||||
int_p: str = parts["int"]
|
||||
ext_p: str = parts["ext"]
|
||||
internal: list[str] = port_range(int_p, parts["int_end"], proto) # type: ignore
|
||||
external = port_range(ext_p or None, parts["ext_end"], "", len(internal) == 1)
|
||||
|
||||
if host is None:
|
||||
if external is not None and len(internal) != len(external):
|
||||
if (external is not None and len(internal) != len(external)) or ext_p == "":
|
||||
raise ValueError("Port ranges don't match in length")
|
||||
return internal, external
|
||||
external_or_none: Sequence[str | None]
|
||||
if not external:
|
||||
external = [None] * len(internal)
|
||||
elif len(internal) != len(external):
|
||||
raise ValueError("Port ranges don't match in length")
|
||||
return internal, [(host, ext_port) for ext_port in external]
|
||||
external_or_none = [None] * len(internal)
|
||||
else:
|
||||
external_or_none = external
|
||||
if len(internal) != len(external_or_none):
|
||||
raise ValueError("Port ranges don't match in length")
|
||||
return internal, [(host, ext_port) for ext_port in external_or_none]
|
||||
|
||||
Reference in New Issue
Block a user