Improve type annotations in core functions

.github/workflows/ci-cd.yml

@@ -113,6 +113,22 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         python -m pip install -e .[ci,dev]
+    # NOTE: Unfortunately the gain of image caching is too small,
+    #       and even it increases the latency of Linux jobs. :(
+    # - name: Cache Docker images
+    #   uses: ScribeMD/[email protected]
+    #   with:
+    #     key: docker-${{ runner.os }}
+    - name: Prepare Docker images (Linux)
+      if: ${{ runner.os == 'Linux' }}
+      run: |
+        docker pull python:3.12-alpine
+    - name: Prepare Docker images (Windows)
+      if: ${{ runner.os == 'Windows' }}
+      # Unfortunately, there is no slim version for Windows.
+      # This may take more than 10 minutes as the image size is a few gigabytes.
+      run: |
+        docker pull python:3.12
     - name: Start Docker services
       if: ${{ matrix.registry == '1' }}
       run: |
@@ -134,6 +150,9 @@ jobs:
         path: coverage-unit-${{ matrix.python-version }}-${{ matrix.os }}-${{ matrix.registry }}.xml
         if-no-files-found: error
         retention-days: 1
+    - name: Clean up Docker images produced during tests
+      run: |
+        docker image list --filter 'reference=aiodocker-*' --format '{{.Repository}}:{{.Tag}}' | xargs -r docker rmi
 
   check:  # This job does nothing and is only used for the branch protection
     name: ✅ Ensure the required checks passing

CHANGES/850.misc

@@ -0,0 +1 @@
+Add more type annotations to the core APIs and retire codes for Python 3.7 compatibility

aiodocker/channel.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 
 

aiodocker/configs.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 import json
 from base64 import b64encode
-from typing import Any, List, Mapping, Optional
+from typing import Any, List, Mapping, Optional, Sequence
 
 from .utils import clean_filters, clean_map
 
@@ -9,7 +11,11 @@ class DockerConfigs:
     def __init__(self, docker):
         self.docker = docker
 
-    async def list(self, *, filters: Optional[Mapping] = None) -> List[Mapping]:
+    async def list(
+        self,
+        *,
+        filters: Optional[Mapping[str, str | Sequence[str]]] = None,
+    ) -> List[Mapping]:
         """
         Return a list of configs
 

aiodocker/containers.py

@@ -1,31 +1,58 @@
+from __future__ import annotations
+
 import json
 import shlex
 import tarfile
-from typing import Any, Dict, Mapping, Optional, Sequence, Tuple, Union
-
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    AsyncIterator,
+    Dict,
+    List,
+    Literal,
+    Mapping,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+    overload,
+)
+
+from aiohttp import ClientWebSocketResponse
 from multidict import MultiDict
 from yarl import URL
 
+from aiodocker.types import JSONObject
+
 from .exceptions import DockerContainerError, DockerError
 from .execs import Exec
 from .jsonstream import json_stream_list, json_stream_stream
 from .logs import DockerLog
 from .multiplexed import multiplexed_result_list, multiplexed_result_stream
 from .stream import Stream
+from .types import PortInfo
 from .utils import identical, parse_result
 
 
+if TYPE_CHECKING:
+    from .docker import Docker
+
+
 class DockerContainers:
-    def __init__(self, docker):
+    def __init__(self, docker: Docker) -> None:
         self.docker = docker
 
-    async def list(self, **kwargs):
+    async def list(self, **kwargs) -> List[DockerContainer]:
         data = await self.docker._query_json(
             "containers/json", method="GET", params=kwargs
         )
         return [DockerContainer(self.docker, **x) for x in data]
 
-    async def create_or_replace(self, name, config):
+    async def create_or_replace(
+        self,
+        name: str,
+        config: JSONObject,
+    ) -> DockerContainer:
         container = None
 
         try:
@@ -44,25 +71,29 @@ async def create_or_replace(self, name, config):
 
         return container
 
-    async def create(self, config, *, name=None):
+    async def create(
+        self,
+        config: JSONObject,
+        *,
+        name: Optional[str] = None,
+    ) -> DockerContainer:
         url = "containers/create"
-
-        config = json.dumps(config, sort_keys=True).encode("utf-8")
+        encoded_config = json.dumps(config, sort_keys=True).encode("utf-8")
         kwargs = {}
         if name:
             kwargs["name"] = name
         data = await self.docker._query_json(
-            url, method="POST", data=config, params=kwargs
+            url, method="POST", data=encoded_config, params=kwargs
         )
         return DockerContainer(self.docker, id=data["Id"])
 
     async def run(
         self,
-        config,
+        config: JSONObject,
         *,
         auth: Optional[Union[Mapping, str, bytes]] = None,
         name: Optional[str] = None,
-    ):
+    ) -> DockerContainer:
         """
         Create and start a container.
 
@@ -77,7 +108,7 @@ async def run(
         except DockerError as err:
             # image not found, try pulling it
             if err.status == 404 and "Image" in config:
-                await self.docker.pull(config["Image"], auth=auth)
+                await self.docker.pull(str(config["Image"]), auth=auth)
                 container = await self.create(config, name=name)
             else:
                 raise err
@@ -91,26 +122,28 @@ async def run(
 
         return container
 
-    async def get(self, container, **kwargs):
+    async def get(self, container_id: str, **kwargs) -> DockerContainer:
         data = await self.docker._query_json(
-            f"containers/{container}/json",
+            f"containers/{container_id}/json",
             method="GET",
             params=kwargs,
         )
         return DockerContainer(self.docker, **data)
 
-    def container(self, container_id, **kwargs):
+    def container(self, container_id: str, **kwargs) -> DockerContainer:
         data = {"id": container_id}
         data.update(kwargs)
         return DockerContainer(self.docker, **data)
 
     def exec(self, exec_id: str) -> Exec:
         """Return Exec instance for already created exec object."""
-        return Exec(self.docker, exec_id, None)
+        return Exec(self.docker, exec_id)
 
 
 class DockerContainer:
-    def __init__(self, docker, **kwargs):
+    _container: Dict[str, Any]
+
+    def __init__(self, docker: Docker, **kwargs) -> None:
         self.docker = docker
         self._container = kwargs
         self._id = self._container.get(
@@ -122,17 +155,42 @@ def __init__(self, docker, **kwargs):
     def id(self) -> str:
         return self._id
 
-    def log(self, *, stdout=False, stderr=False, follow=False, **kwargs):
+    @overload
+    async def log(
+        self,
+        *,
+        stdout: bool = False,
+        stderr: bool = False,
+        follow: Literal[False] = False,
+        **kwargs,
+    ) -> List[str]: ...
+
+    @overload
+    def log(
+        self,
+        *,
+        stdout: bool = False,
+        stderr: bool = False,
+        follow: Literal[True],
+        **kwargs,
+    ) -> AsyncIterator[str]: ...
+
+    def log(
+        self,
+        *,
+        stdout: bool = False,
+        stderr: bool = False,
+        follow: bool = False,
+        **kwargs,
+    ) -> Any:
         if stdout is False and stderr is False:
             raise TypeError("Need one of stdout or stderr")
 
         params = {"stdout": stdout, "stderr": stderr, "follow": follow}
         params.update(kwargs)
-
         cm = self.docker._query(
             f"containers/{self._id}/logs", method="GET", params=params
         )
-
         if follow:
             return self._logs_stream(cm)
         else:
@@ -154,7 +212,6 @@ async def _logs_list(self, cm):
         try:
             inspect_info = await self.show()
         except DockerError:
-            cm.cancel()
             raise
         is_tty = inspect_info["Config"]["Tty"]
 
@@ -181,20 +238,20 @@ async def put_archive(self, path, data):
             data = await parse_result(response)
             return data
 
-    async def show(self, **kwargs):
+    async def show(self, **kwargs) -> Dict[str, Any]:
         data = await self.docker._query_json(
             f"containers/{self._id}/json", method="GET", params=kwargs
         )
         self._container = data
         return data
 
-    async def stop(self, **kwargs):
+    async def stop(self, **kwargs) -> None:
         async with self.docker._query(
             f"containers/{self._id}/stop", method="POST", params=kwargs
         ):
             pass
 
-    async def start(self, **kwargs):
+    async def start(self, **kwargs) -> None:
         async with self.docker._query(
             f"containers/{self._id}/start",
             method="POST",
@@ -203,7 +260,7 @@ async def start(self, **kwargs):
         ):
             pass
 
-    async def restart(self, timeout=None):
+    async def restart(self, timeout=None) -> None:
         params = {}
         if timeout is not None:
             params["t"] = timeout
@@ -214,13 +271,13 @@ async def restart(self, timeout=None):
         ):
             pass
 
-    async def kill(self, **kwargs):
+    async def kill(self, **kwargs) -> None:
         async with self.docker._query(
             f"containers/{self._id}/kill", method="POST", params=kwargs
         ):
             pass
 
-    async def wait(self, *, timeout=None, **kwargs):
+    async def wait(self, *, timeout=None, **kwargs) -> Dict[str, Any]:
         data = await self.docker._query_json(
             f"containers/{self._id}/wait",
             method="POST",
@@ -229,13 +286,13 @@ async def wait(self, *, timeout=None, **kwargs):
         )
         return data
 
-    async def delete(self, **kwargs):
+    async def delete(self, **kwargs) -> None:
         async with self.docker._query(
             f"containers/{self._id}", method="DELETE", params=kwargs
         ):
             pass
 
-    async def rename(self, newname):
+    async def rename(self, newname) -> None:
         async with self.docker._query(
             f"containers/{self._id}/rename",
             method="POST",
@@ -244,7 +301,7 @@ async def rename(self, newname):
         ):
             pass
 
-    async def websocket(self, **params):
+    async def websocket(self, **params) -> ClientWebSocketResponse:
         if not params:
             params = {"stdin": True, "stdout": True, "stderr": True, "stream": True}
         path = f"containers/{self._id}/attach/ws"
@@ -280,7 +337,7 @@ async def setup() -> Tuple[URL, Optional[bytes], bool]:
 
         return Stream(self.docker, setup, None)
 
-    async def port(self, private_port):
+    async def port(self, private_port: int | str) -> List[PortInfo] | None:
         if "NetworkSettings" not in self._container:
             await self.show()
 
@@ -302,7 +359,25 @@ async def port(self, private_port):
 
         return h_ports
 
-    def stats(self, *, stream=True):
+    @overload
+    def stats(
+        self,
+        *,
+        stream: Literal[True] = True,
+    ) -> AsyncIterator[Dict[str, Any]]: ...
+
+    @overload
+    async def stats(
+        self,
+        *,
+        stream: Literal[False],
+    ) -> List[Dict[str, Any]]: ...
+
+    def stats(
+        self,
+        *,
+        stream: bool = True,
+    ) -> Any:
         cm = self.docker._query(
             f"containers/{self._id}/stats",
             params={"stream": "1" if stream else "0"},
@@ -333,7 +408,7 @@ async def exec(
         environment: Optional[Union[Mapping[str, str], Sequence[str]]] = None,
         workdir: Optional[str] = None,
         detach_keys: Optional[str] = None,
-    ):
+    ) -> Exec:
         if isinstance(cmd, str):
             cmd = shlex.split(cmd)
         if environment is None:
@@ -418,8 +493,8 @@ async def unpause(self) -> None:
         async with self.docker._query(f"containers/{self._id}/unpause", method="POST"):
             pass
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: str) -> Any:
         return self._container[key]
 
-    def __hasitem__(self, key):
+    def __hasitem__(self, key: str) -> bool:
         return key in self._container