feat(container): move docker response dataclasses to docker_client.py

Author: VictorCavichioli

core/testcontainers/compose/compose.py

@@ -1,5 +1,5 @@
 import sys
-from dataclasses import asdict, dataclass, field, fields, is_dataclass
+from dataclasses import asdict, dataclass, field
 from functools import cached_property
 from json import loads
 from logging import getLogger, warning
@@ -11,6 +11,7 @@
 from types import TracebackType
 from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast
 
+from testcontainers.core.docker_client import ContainerInspectInfo, _ignore_properties
 from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed
 from testcontainers.core.waiting_utils import WaitStrategy
 
@@ -20,135 +21,6 @@
 logger = getLogger(__name__)
 
 
-def _ignore_properties(cls: type[_IPT], dict_: Any) -> _IPT:
-    """omits extra fields like @JsonIgnoreProperties(ignoreUnknown = true)
-
-    https://gist.github.com/alexanderankin/2a4549ac03554a31bef6eaaf2eaf7fd5"""
-    if isinstance(dict_, cls):
-        return dict_
-    if not is_dataclass(cls):
-        raise TypeError(f"Expected a dataclass type, got {cls}")
-    class_fields = {f.name for f in fields(cls)}
-    filtered = {k: v for k, v in dict_.items() if k in class_fields}
-    return cast("_IPT", cls(**filtered))
-
-
-@dataclass
-class ContainerState:
-    """Container state from docker inspect."""
-
-    Status: Optional[str] = None
-    Running: Optional[bool] = None
-    Paused: Optional[bool] = None
-    Restarting: Optional[bool] = None
-    OOMKilled: Optional[bool] = None
-    Dead: Optional[bool] = None
-    Pid: Optional[int] = None
-    ExitCode: Optional[int] = None
-    Error: Optional[str] = None
-    StartedAt: Optional[str] = None
-    FinishedAt: Optional[str] = None
-
-
-@dataclass
-class ContainerConfig:
-    """Container config from docker inspect."""
-
-    Hostname: Optional[str] = None
-    User: Optional[str] = None
-    Env: Optional[list[str]] = None
-    Cmd: Optional[list[str]] = None
-    Image: Optional[str] = None
-    WorkingDir: Optional[str] = None
-    Entrypoint: Optional[list[str]] = None
-    ExposedPorts: Optional[dict[str, Any]] = None
-    Labels: Optional[dict[str, str]] = None
-
-
-@dataclass
-class Network:
-    """Individual network from docker inspect."""
-
-    IPAddress: Optional[str] = None
-    Gateway: Optional[str] = None
-    NetworkID: Optional[str] = None
-    EndpointID: Optional[str] = None
-    MacAddress: Optional[str] = None
-    Aliases: Optional[list[str]] = None
-
-
-@dataclass
-class NetworkSettings:
-    """Network settings from docker inspect."""
-
-    Bridge: Optional[str] = None
-    IPAddress: Optional[str] = None
-    Gateway: Optional[str] = None
-    Ports: Optional[dict[str, Any]] = None
-    Networks: Optional[dict[str, Network]] = None
-
-    def get_networks(self) -> Optional[dict[str, Network]]:
-        """Get networks for the container."""
-        return self.Networks
-
-
-@dataclass
-class ContainerInspectInfo:
-    """Container information from docker inspect."""
-
-    Id: Optional[str] = None
-    Name: Optional[str] = None
-    Created: Optional[str] = None
-    Path: Optional[str] = None
-    Args: Optional[list[str]] = None
-    Image: Optional[str] = None
-    State: Optional[ContainerState] = None
-    Config: Optional[ContainerConfig] = None
-    network_settings: Optional[NetworkSettings] = None
-    Mounts: Optional[list[dict[str, Any]]] = None
-
-    @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> "ContainerInspectInfo":
-        """Create from docker inspect JSON."""
-        return cls(
-            Id=data.get("Id"),
-            Name=data.get("Name"),
-            Created=data.get("Created"),
-            Path=data.get("Path"),
-            Args=data.get("Args"),
-            Image=data.get("Image"),
-            State=_ignore_properties(ContainerState, data.get("State", {})) if data.get("State") else None,
-            Config=_ignore_properties(ContainerConfig, data.get("Config", {})) if data.get("Config") else None,
-            network_settings=cls._parse_network_settings(data.get("NetworkSettings", {}))
-            if data.get("NetworkSettings")
-            else None,
-            Mounts=data.get("Mounts"),
-        )
-
-    @classmethod
-    def _parse_network_settings(cls, data: dict[str, Any]) -> Optional[NetworkSettings]:
-        """Parse NetworkSettings with Networks as Network objects."""
-        if not data:
-            return None
-
-        networks_data = data.get("Networks", {})
-        networks = {}
-        for name, net_data in networks_data.items():
-            networks[name] = _ignore_properties(Network, net_data)
-
-        return NetworkSettings(
-            Bridge=data.get("Bridge"),
-            IPAddress=data.get("IPAddress"),
-            Gateway=data.get("Gateway"),
-            Ports=data.get("Ports"),
-            Networks=networks,
-        )
-
-    def get_network_settings(self) -> Optional[NetworkSettings]:
-        """Get network settings for the container."""
-        return self.network_settings
-
-
 @dataclass
 class PublishedPortModel:
     """

core/testcontainers/core/container.py

@@ -14,7 +14,7 @@
 
 from testcontainers.core.config import ConnectionMode
 from testcontainers.core.config import testcontainers_config as c
-from testcontainers.core.docker_client import DockerClient
+from testcontainers.core.docker_client import ContainerInspectInfo, DockerClient
 from testcontainers.core.exceptions import ContainerConnectException, ContainerStartException
 from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID
 from testcontainers.core.network import Network
@@ -96,6 +96,7 @@ def __init__(
 
         self._kwargs = kwargs
         self._wait_strategy: Optional[WaitStrategy] = _wait_strategy
+        self._cached_container_info: Optional[ContainerInspectInfo] = None
 
     def with_env(self, key: str, value: str) -> Self:
         self.env[key] = value
@@ -300,6 +301,24 @@ def exec(self, command: Union[str, list[str]]) -> ExecResult:
             raise ContainerStartException("Container should be started before executing a command")
         return self._container.exec_run(command)
 
+    def get_container_info(self) -> Optional[ContainerInspectInfo]:
+        """Get container information via docker inspect (lazy loaded)."""
+        if self._cached_container_info is not None:
+            return self._cached_container_info
+
+        if not self._container:
+            return None
+
+        try:
+            raw_data = self._container.attrs
+            self._cached_container_info = ContainerInspectInfo.from_dict(raw_data)
+
+        except Exception as e:
+            logger.warning(f"Failed to get container info for {self._container.id}: {e}")
+            self._cached_container_info = None
+
+        return self._cached_container_info
+
     def _configure(self) -> None:
         # placeholder if subclasses want to define this and use the default start method
         pass