fix: typing in core

.pre-commit-config.yaml

@@ -22,13 +22,13 @@ repos:
         # Explicitly setting config to prevent Ruff from using `pyproject.toml` in sub packages.
         args: [ '--fix', '--exit-non-zero-on-fix', '--config', 'pyproject.toml' ]
 
-#  - repo: local
-#    hooks:
-#      - id: mypy
-#        name: mypy
-#        entry: poetry run mypy
-#        args: ["--config-file", "pyproject.toml"]
-#        files: "core"  # start with the core being type checked
-#        language: system
-#        types: [ python ]
-#        require_serial: true
+  - repo: local
+    hooks:
+      - id: mypy
+        name: mypy
+        entry: poetry run mypy
+        args: ["--config-file", "pyproject.toml"]
+        files: "core"  # start with the core being type checked
+        language: system
+        types: [ python ]
+        require_serial: true

core/testcontainers/compose/__init__.py

@@ -6,3 +6,11 @@
     ComposeContainer,
     DockerCompose,
 )
+
+__all__ = [
+    "ContainerIsNotRunning",
+    "NoSuchPortExposed",
+    "PublishedPort",
+    "ComposeContainer",
+    "DockerCompose",
+]

core/testcontainers/compose/compose.py

@@ -1,11 +1,11 @@
-from dataclasses import dataclass, field, fields
+from dataclasses import Field, dataclass, field, fields
 from functools import cached_property
 from json import loads
 from os import PathLike
 from re import split
 from subprocess import CompletedProcess
 from subprocess import run as subprocess_run
-from typing import Callable, Literal, Optional, TypeVar, Union
+from typing import Any, Callable, ClassVar, Literal, Optional, Protocol, TypeVar, Union, cast
 from urllib.error import HTTPError, URLError
 from urllib.request import urlopen
 
@@ -15,14 +15,18 @@
 _IPT = TypeVar("_IPT")
 
 
-def _ignore_properties(cls: type[_IPT], dict_: any) -> _IPT:
+class DataclassInstance(Protocol):
+    __dataclass_fields__: ClassVar[dict[str, Field[Any]]]
+
+
+def _ignore_properties(cls: type[_IPT], dict_: Union[_IPT, dict[str, Any]]) -> _IPT:
     """omits extra fields like @JsonIgnoreProperties(ignoreUnknown = true)
 
     https://gist.github.com/alexanderankin/2a4549ac03554a31bef6eaaf2eaf7fd5"""
     if isinstance(dict_, cls):
         return dict_
-    class_fields = {f.name for f in fields(cls)}
-    filtered = {k: v for k, v in dict_.items() if k in class_fields}
+    class_fields = {f.name for f in fields(cast(type[DataclassInstance], cls))}
+    filtered = {k: v for k, v in cast(dict[str, Any], dict_).items() if k in class_fields}
     return cls(**filtered)
 
 
@@ -34,8 +38,8 @@ class PublishedPort:
     """
 
     URL: Optional[str] = None
-    TargetPort: Optional[str] = None
-    PublishedPort: Optional[str] = None
+    TargetPort: Optional[int] = None
+    PublishedPort: Optional[int] = None
     Protocol: Optional[str] = None
 
 
@@ -61,21 +65,21 @@ class ComposeContainer:
     Name: Optional[str] = None
     Command: Optional[str] = None
     Project: Optional[str] = None
-    Service: Optional[str] = None
+    Service: str = ""
     State: Optional[str] = None
     Health: Optional[str] = None
-    ExitCode: Optional[str] = None
+    ExitCode: Optional[int] = None
     Publishers: list[PublishedPort] = field(default_factory=list)
 
-    def __post_init__(self):
+    def __post_init__(self) -> None:
         if self.Publishers:
             self.Publishers = [_ignore_properties(PublishedPort, p) for p in self.Publishers]
 
     def get_publisher(
         self,
         by_port: Optional[int] = None,
         by_host: Optional[str] = None,
-        prefer_ip_version: Literal["IPV4", "IPv6"] = "IPv4",
+        prefer_ip_version: Literal["IPv4", "IPv6"] = "IPv4",
     ) -> PublishedPort:
         remaining_publishers = self.Publishers
 
@@ -98,8 +102,8 @@ def get_publisher(
         )
 
     @staticmethod
-    def _matches_protocol(prefer_ip_version, r):
-        return (":" in r.URL) is (prefer_ip_version == "IPv6")
+    def _matches_protocol(prefer_ip_version: str, r: PublishedPort) -> bool:
+        return (":" in (r.URL or "")) is (prefer_ip_version == "IPv6")
 
 
 @dataclass
@@ -151,23 +155,23 @@ class DockerCompose:
                 image: "hello-world"
     """
 
-    context: Union[str, PathLike]
+    context: Union[str, PathLike[Any]]
     compose_file_name: Optional[Union[str, list[str]]] = None
     pull: bool = False
     build: bool = False
     wait: bool = True
     env_file: Optional[str] = None
     services: Optional[list[str]] = None
 
-    def __post_init__(self):
+    def __post_init__(self) -> None:
         if isinstance(self.compose_file_name, str):
             self.compose_file_name = [self.compose_file_name]
 
     def __enter__(self) -> "DockerCompose":
         self.start()
         return self
 
-    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         self.stop()
 
     def docker_compose_command(self) -> list[str]:
@@ -217,7 +221,7 @@ def start(self) -> None:
 
         self._run_command(cmd=up_cmd)
 
-    def stop(self, down=True) -> None:
+    def stop(self, down: bool = True) -> None:
         """
         Stops the docker compose environment.
         """
@@ -243,7 +247,7 @@ def get_logs(self, *services: str) -> tuple[str, str]:
         result = self._run_command(cmd=logs_cmd)
         return result.stdout.decode("utf-8"), result.stderr.decode("utf-8")
 
-    def get_containers(self, include_all=False) -> list[ComposeContainer]:
+    def get_containers(self, include_all: bool = False) -> list[ComposeContainer]:
         """
         Fetch information about running containers via `docker compose ps --format json`.
         Available only in V2 of compose.
@@ -326,7 +330,7 @@ def exec_in_container(
     def _run_command(
         self,
         cmd: Union[str, list[str]],
-        context: Optional[str] = None,
+        context: Optional[Union[str, PathLike[Any]]] = None,
     ) -> CompletedProcess[bytes]:
         context = context or self.context
         return subprocess_run(
@@ -340,7 +344,7 @@ def get_service_port(
         self,
         service_name: Optional[str] = None,
         port: Optional[int] = None,
-    ):
+    ) -> Optional[int]:
         """
         Returns the mapped port for one of the services.
 
@@ -362,7 +366,7 @@ def get_service_host(
         self,
         service_name: Optional[str] = None,
         port: Optional[int] = None,
-    ):
+    ) -> Optional[str]:
         """
         Returns the host for one of the services.
 
@@ -384,7 +388,7 @@ def get_service_host_and_port(
         self,
         service_name: Optional[str] = None,
         port: Optional[int] = None,
-    ):
+    ) -> tuple[Optional[str], Optional[int]]:
         publisher = self.get_container(service_name).get_publisher(by_port=port)
         return publisher.URL, publisher.PublishedPort
 
@@ -401,3 +405,12 @@ def wait_for(self, url: str) -> "DockerCompose":
         with urlopen(url) as response:
             response.read()
         return self
+
+
+__all__ = [
+    "ContainerIsNotRunning",
+    "NoSuchPortExposed",
+    "PublishedPort",
+    "ComposeContainer",
+    "DockerCompose",
+]

core/testcontainers/core/container.py

@@ -1,6 +1,7 @@
+from os import PathLike
 from platform import system
 from socket import socket
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional, TypedDict, Union
 
 from testcontainers.core.config import RYUK_DISABLED, RYUK_DOCKER_SOCKET, RYUK_IMAGE, RYUK_PRIVILEGED
 from testcontainers.core.docker_client import DockerClient
@@ -10,11 +11,16 @@
 from testcontainers.core.waiting_utils import wait_container_is_ready, wait_for_logs
 
 if TYPE_CHECKING:
-    from docker.models.containers import Container
+    from docker.models.containers import Container, ExecResult
 
 logger = setup_logger(__name__)
 
 
+class VolumeDict(TypedDict):
+    bind: str
+    mode: str
+
+
 class DockerContainer:
     """
     Basic container object to spin up Docker instances.
@@ -31,19 +37,26 @@ class DockerContainer:
     def __init__(
         self,
         image: str,
-        docker_client_kw: Optional[dict] = None,
-        **kwargs,
+        docker_client_kw: Optional[dict[str, Any]] = None,
+        **kwargs: Any,
     ) -> None:
-        self.env = {}
-        self.ports = {}
-        self.volumes = {}
+        self.env: dict[str, str] = {}
+        self.ports: dict[int, Optional[int]] = {}
+        self.volumes: dict[str, VolumeDict] = {}
         self.image = image
         self._docker = DockerClient(**(docker_client_kw or {}))
-        self._container = None
-        self._command = None
-        self._name = None
+        self._container: Optional["Container"] = None
+        self._command: Union[str, list[str], None] = None
+        self._name: Optional[str] = None
         self._kwargs = kwargs
 
+    @property
+    def _use_container(self) -> "Container":
+        """todo fail fast with better error when it is None"""
+        container = self._container
+        assert container is not None
+        return container
+
     def with_env(self, key: str, value: str) -> "DockerContainer":
         self.env[key] = value
         return self
@@ -57,7 +70,7 @@ def with_exposed_ports(self, *ports: int) -> "DockerContainer":
             self.ports[port] = None
         return self
 
-    def with_kwargs(self, **kwargs) -> "DockerContainer":
+    def with_kwargs(self, **kwargs: Any) -> "DockerContainer":
         self._kwargs = kwargs
         return self
 
@@ -66,7 +79,7 @@ def maybe_emulate_amd64(self) -> "DockerContainer":
             return self.with_kwargs(platform="linux/amd64")
         return self
 
-    def start(self):
+    def start(self) -> "DockerContainer":
         if not RYUK_DISABLED and self.image != RYUK_IMAGE:
             logger.debug("Creating Ryuk container")
             Reaper.get_instance()
@@ -85,14 +98,14 @@ def start(self):
         logger.info("Container started: %s", self._container.short_id)
         return self
 
-    def stop(self, force=True, delete_volume=True) -> None:
-        self._container.remove(force=force, v=delete_volume)
+    def stop(self, force: bool = True, delete_volume: bool = True) -> None:
+        self._use_container.remove(force=force, v=delete_volume)
         self.get_docker_client().client.close()
 
-    def __enter__(self):
+    def __enter__(self) -> "DockerContainer":
         return self.start()
 
-    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         self.stop()
 
     def get_container_host_ip(self) -> str:
@@ -120,13 +133,13 @@ def get_container_host_ip(self) -> str:
 
     @wait_container_is_ready()
     def get_exposed_port(self, port: int) -> str:
-        mapped_port = self.get_docker_client().port(self._container.id, port)
+        mapped_port = self.get_docker_client().port(self._use_container.id, port)
         if inside_container():
-            gateway_ip = self.get_docker_client().gateway_ip(self._container.id)
+            gateway_ip = self.get_docker_client().gateway_ip(self._use_container.id)
             host = self.get_docker_client().host()
 
             if gateway_ip == host:
-                return port
+                return str(port)
         return mapped_port
 
     def with_command(self, command: str) -> "DockerContainer":
@@ -137,9 +150,11 @@ def with_name(self, name: str) -> "DockerContainer":
         self._name = name
         return self
 
-    def with_volume_mapping(self, host: str, container: str, mode: str = "ro") -> "DockerContainer":
-        mapping = {"bind": container, "mode": mode}
-        self.volumes[host] = mapping
+    def with_volume_mapping(
+        self, host: Union[str, PathLike[Any]], container: str, mode: str = "ro"
+    ) -> "DockerContainer":
+        mapping: VolumeDict = {"bind": container, "mode": mode}
+        self.volumes[str(host)] = mapping
         return self
 
     def get_wrapped_container(self) -> "Container":
@@ -153,7 +168,7 @@ def get_logs(self) -> tuple[bytes, bytes]:
             raise ContainerStartException("Container should be started before getting logs")
         return self._container.logs(stderr=False), self._container.logs(stdout=False)
 
-    def exec(self, command) -> tuple[int, str]:
+    def exec(self, command: Union[str, list[str]]) -> "ExecResult":
         if not self._container:
             raise ContainerStartException("Container should be started before executing a command")
         return self._container.exec_run(command)