Improve types in docker models (#11859)

stubs/docker/@tests/stubtest_allowlist.txt

@@ -5,3 +5,6 @@ docker.transport.SSHHTTPAdapter
 docker.transport.npipeconn
 docker.transport.npipesocket
 docker.transport.sshconn
+
+# model is always set by child classes
+docker.models.resource.Collection.model

stubs/docker/docker/models/configs.pyi

@@ -3,11 +3,11 @@ from .resource import Collection, Model
 class Config(Model):
     id_attribute: str
     @property
-    def name(self): ...
-    def remove(self): ...
+    def name(self) -> str: ...
+    def remove(self) -> bool: ...
 
-class ConfigCollection(Collection):
+class ConfigCollection(Collection[Config]):
     model: type[Config]
-    def create(self, **kwargs): ...
-    def get(self, config_id): ...
-    def list(self, **kwargs): ...
+    def create(self, **kwargs) -> Config: ...  # type:ignore[override]
+    def get(self, config_id: str) -> Config: ...
+    def list(self, **kwargs) -> list[Config]: ...

stubs/docker/docker/models/containers.pyi

@@ -54,7 +54,7 @@ class Container(Model):
     def update(self, **kwargs): ...
     def wait(self, **kwargs): ...
 
-class ContainerCollection(Collection):
+class ContainerCollection(Collection[Container]):
     model: type[Container]
     def run(
         self, image, command: Incomplete | None = None, stdout: bool = True, stderr: bool = False, remove: bool = False, **kwargs

stubs/docker/docker/models/images.pyi

@@ -1,45 +1,51 @@
-from _typeshed import Incomplete
-from collections.abc import Generator
-from typing import Any
+from collections.abc import Iterator
+from typing import Any, Literal, overload
+from typing_extensions import TypeAlias
 
 from .resource import Collection, Model
 
+_ImageList: TypeAlias = list[Image]  # To resolve conflicts with a method called "list"
+
 class Image(Model):
     @property
-    def labels(self): ...
+    def labels(self) -> dict[str, Any]: ...
     @property
-    def short_id(self): ...
+    def short_id(self) -> str: ...
     @property
-    def tags(self): ...
-    def history(self): ...
-    def remove(self, force: bool = False, noprune: bool = False): ...
-    def save(self, chunk_size=2097152, named: bool = False): ...
-    def tag(self, repository, tag: str | None = None, **kwargs): ...
+    def tags(self) -> list[str]: ...
+    def history(self) -> list[Any]: ...
+    def remove(self, force: bool = False, noprune: bool = False) -> dict[str, Any]: ...
+    def save(self, chunk_size: int = 2097152, named: bool = False) -> Iterator[Any]: ...
+    def tag(self, repository: str, tag: str | None = None, **kwargs) -> bool: ...
 
 class RegistryData(Model):
-    image_name: Incomplete
-    def __init__(self, image_name, *args, **kwargs) -> None: ...
+    image_name: str
+    def __init__(self, image_name: str, *args, **kwargs) -> None: ...
     @property
-    def id(self): ...
+    def id(self) -> str: ...
     @property
-    def short_id(self): ...
-    def pull(self, platform: Incomplete | None = None): ...
+    def short_id(self) -> str: ...
+    def pull(self, platform: str | None = None) -> Image: ...
     def has_platform(self, platform): ...
-    attrs: Incomplete
     def reload(self) -> None: ...
 
-class ImageCollection(Collection):
+class ImageCollection(Collection[Image]):
     model: type[Image]
-    def build(self, **kwargs) -> tuple[Image, Generator[Any, None, None]]: ...
-    def get(self, name): ...
-    def get_registry_data(self, name, auth_config: Incomplete | None = None): ...
-    def list(self, name: Incomplete | None = None, all: bool = False, filters: Incomplete | None = None): ...
-    def load(self, data): ...
-    def pull(self, repository, tag: str | None = None, all_tags: bool = False, **kwargs): ...
-    def push(self, repository, tag: str | None = None, **kwargs): ...
+    def build(self, **kwargs) -> tuple[Image, Iterator[Any]]: ...
+    def get(self, name: str) -> Image: ...
+    def get_registry_data(self, name, auth_config: dict[str, Any] | None = None) -> RegistryData: ...
+    def list(self, name: str | None = None, all: bool = False, filters: dict[str, Any] | None = None) -> _ImageList: ...
+    def load(self, data: bytes) -> _ImageList: ...
+    @overload
+    def pull(self, repository: str, tag: str | None = None, all_tags: Literal[False] = False, **kwargs) -> Image: ...
+    @overload
+    def pull(self, repository: str, tag: str | None = None, *, all_tags: Literal[True], **kwargs) -> _ImageList: ...
+    @overload
+    def pull(self, repository: str, tag: str | None, all_tags: Literal[True], **kwargs) -> _ImageList: ...
+    def push(self, repository: str, tag: str | None = None, **kwargs): ...
     def remove(self, *args, **kwargs) -> None: ...
     def search(self, *args, **kwargs): ...
-    def prune(self, filters: Incomplete | None = None): ...
+    def prune(self, filters: dict[str, Any] | None = None): ...
     def prune_builds(self, *args, **kwargs): ...
 
 def normalize_platform(platform, engine_info): ...

stubs/docker/docker/models/networks.pyi

@@ -1,19 +1,20 @@
-from _typeshed import Incomplete
+from typing import Any
 
+from .containers import Container
 from .resource import Collection, Model
 
 class Network(Model):
     @property
-    def name(self): ...
+    def name(self) -> str | None: ...
     @property
-    def containers(self): ...
-    def connect(self, container, *args, **kwargs): ...
-    def disconnect(self, container, *args, **kwargs): ...
-    def remove(self): ...
+    def containers(self) -> list[Container]: ...
+    def connect(self, container: str | Container, *args, **kwargs) -> None: ...
+    def disconnect(self, container: str | Container, *args, **kwargs) -> None: ...
+    def remove(self) -> None: ...
 
-class NetworkCollection(Collection):
+class NetworkCollection(Collection[Network]):
     model: type[Network]
-    def create(self, name, *args, **kwargs): ...
-    def get(self, network_id, *args, **kwargs): ...
-    def list(self, *args, **kwargs): ...
-    def prune(self, filters: Incomplete | None = None): ...
+    def create(self, name: str, *args, **kwargs) -> Network: ...  # type:ignore[override]
+    def get(self, network_id: str, *args, **kwargs) -> Network: ...  # type:ignore[override]
+    def list(self, *args, **kwargs) -> list[Network]: ...
+    def prune(self, filters: dict[str, Any] | None = None) -> dict[str, Any]: ...

stubs/docker/docker/models/nodes.pyi

@@ -7,7 +7,7 @@ class Node(Model):
     def update(self, node_spec): ...
     def remove(self, force: bool = False): ...
 
-class NodeCollection(Collection):
+class NodeCollection(Collection[Node]):
     model: type[Node]
     def get(self, node_id): ...
     def list(self, *args, **kwargs): ...

stubs/docker/docker/models/plugins.pyi

@@ -17,7 +17,7 @@ class Plugin(Model):
     def remove(self, force: bool = False): ...
     def upgrade(self, remote: Incomplete | None = None) -> Generator[Incomplete, Incomplete, None]: ...
 
-class PluginCollection(Collection):
+class PluginCollection(Collection[Plugin]):
     model: type[Plugin]
     def create(self, name, plugin_data_dir, gzip: bool = False): ...  # type:ignore[override]
     def get(self, name): ...

stubs/docker/docker/models/resource.pyi

@@ -1,27 +1,32 @@
-from _typeshed import Incomplete
+from typing import Any, Generic, NoReturn, TypeVar
+from typing_extensions import Self
+
+from docker import APIClient
+
+_T = TypeVar("_T", bound=Model)
 
 class Model:
     id_attribute: str
-    client: Incomplete
-    collection: Incomplete
-    attrs: Incomplete
+    client: APIClient | None
+    collection: Collection[Self] | None
+    attrs: dict[str, Any]
     def __init__(
-        self, attrs: Incomplete | None = None, client: Incomplete | None = None, collection: Incomplete | None = None
+        self, attrs: dict[str, Any] | None = None, client: APIClient | None = None, collection: Collection[Self] | None = None
     ) -> None: ...
-    def __eq__(self, other): ...
-    def __hash__(self): ...
+    def __eq__(self, other) -> bool: ...
+    def __hash__(self) -> int: ...
     @property
-    def id(self): ...
+    def id(self) -> str | None: ...
     @property
-    def short_id(self): ...
+    def short_id(self) -> str: ...
     def reload(self) -> None: ...
 
-class Collection:
-    model: Incomplete
-    client: Incomplete
-    def __init__(self, client: Incomplete | None = None) -> None: ...
-    def __call__(self, *args, **kwargs) -> None: ...
-    def list(self) -> None: ...
-    def get(self, key) -> None: ...
-    def create(self, attrs: Incomplete | None = None) -> None: ...
-    def prepare_model(self, attrs): ...
+class Collection(Generic[_T]):
+    model: type[_T]
+    client: APIClient
+    def __init__(self, client: APIClient | None = None) -> None: ...
+    def __call__(self, *args, **kwargs) -> NoReturn: ...
+    def list(self) -> list[_T]: ...
+    def get(self, key: str) -> _T: ...
+    def create(self, attrs: Any | None = None) -> _T: ...
+    def prepare_model(self, attrs: Model | dict[str, Any]) -> _T: ...

stubs/docker/docker/models/secrets.pyi

@@ -6,7 +6,7 @@ class Secret(Model):
     def name(self): ...
     def remove(self): ...
 
-class SecretCollection(Collection):
+class SecretCollection(Collection[Secret]):
     model: type[Secret]
     def create(self, **kwargs): ...
     def get(self, secret_id): ...

stubs/docker/docker/models/services.pyi

@@ -15,7 +15,7 @@ class Service(Model):
     def scale(self, replicas): ...
     def force_update(self): ...
 
-class ServiceCollection(Collection):
+class ServiceCollection(Collection[Service]):
     model: type[Service]
     def create(self, image, command: Incomplete | None = None, **kwargs): ...  # type:ignore[override]
     def get(self, service_id, insert_defaults: Incomplete | None = None): ...

stubs/docker/docker/models/swarm.pyi

@@ -1,33 +1,33 @@
-from _typeshed import Incomplete
+from collections.abc import Iterable
+from typing import Any
 
 from .resource import Model
 
 class Swarm(Model):
     id_attribute: str
     def __init__(self, *args, **kwargs) -> None: ...
     @property
-    def version(self): ...
-    def get_unlock_key(self): ...
+    def version(self) -> str | None: ...
+    def get_unlock_key(self) -> dict[str, Any]: ...
     def init(
         self,
-        advertise_addr: Incomplete | None = None,
+        advertise_addr: str | None = None,
         listen_addr: str = "0.0.0.0:2377",
         force_new_cluster: bool = False,
-        default_addr_pool: Incomplete | None = None,
-        subnet_size: Incomplete | None = None,
-        data_path_addr: Incomplete | None = None,
-        data_path_port: Incomplete | None = None,
+        default_addr_pool: Iterable[str] | None = None,
+        subnet_size: int | None = None,
+        data_path_addr: str | None = None,
+        data_path_port: int | None = None,
         **kwargs,
-    ): ...
-    def join(self, *args, **kwargs): ...
-    def leave(self, *args, **kwargs): ...
-    attrs: Incomplete
+    ) -> str: ...
+    def join(self, *args, **kwargs) -> bool: ...
+    def leave(self, *args, **kwargs) -> bool: ...
     def reload(self) -> None: ...
-    def unlock(self, key): ...
+    def unlock(self, key: str) -> bool: ...
     def update(
         self,
         rotate_worker_token: bool = False,
         rotate_manager_token: bool = False,
         rotate_manager_unlock_key: bool = False,
         **kwargs,
-    ): ...
+    ) -> bool: ...

stubs/docker/docker/models/volumes.pyi

@@ -1,16 +1,16 @@
-from _typeshed import Incomplete
+from typing import Any
 
 from .resource import Collection, Model
 
 class Volume(Model):
     id_attribute: str
     @property
-    def name(self): ...
-    def remove(self, force: bool = False): ...
+    def name(self) -> str: ...
+    def remove(self, force: bool = False) -> None: ...
 
-class VolumeCollection(Collection):
+class VolumeCollection(Collection[Volume]):
     model: type[Volume]
-    def create(self, name: Incomplete | None = None, **kwargs): ...  # type:ignore[override]
-    def get(self, volume_id): ...
-    def list(self, **kwargs): ...
-    def prune(self, filters: Incomplete | None = None): ...
+    def create(self, name: str | None = None, **kwargs) -> Volume: ...  # type:ignore[override]
+    def get(self, volume_id: str) -> Volume: ...
+    def list(self, **kwargs) -> list[Volume]: ...
+    def prune(self, filters: dict[str, Any] | None = None) -> dict[str, Any]: ...