feat(compose): add structured container inspect information (#897)

Summary
This PR adds the ability to retrieve detailed container information from
docker inspect in a structured format for ComposeContainer objects, with
lazy loading and caching for optimal performance.

Related with #857

---------

Co-authored-by: Roy Moore <roy@moore.co.il>

Author: VictorCavichioli

conf.py

@@ -168,4 +168,5 @@
 nitpick_ignore = [
     ("py:class", "typing_extensions.Self"),
     ("py:class", "docker.models.containers.ExecResult"),
+    ("py:class", "testcontainers.core.docker_client.ContainerInspectInfo"),
 ]

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,29 +11,16 @@
 from types import TracebackType
 from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast
 
-from testcontainers.core.docker_client import get_docker_host_hostname
+from testcontainers.core.docker_client import DockerClient, get_docker_host_hostname
 from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed
+from testcontainers.core.inspect import ContainerInspectInfo, _ignore_properties
 from testcontainers.core.waiting_utils import WaitStrategy
 
-_IPT = TypeVar("_IPT")
 _WARNINGS = {"DOCKER_COMPOSE_GET_CONFIG": "get_config is experimental, see testcontainers/testcontainers-python#669"}
 
 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 cls(**filtered)
-
-
 @dataclass
 class PublishedPortModel:
     """
@@ -93,6 +80,7 @@ class ComposeContainer:
     ExitCode: Optional[int] = None
     Publishers: list[PublishedPortModel] = field(default_factory=list)
     _docker_compose: Optional["DockerCompose"] = field(default=None, init=False, repr=False)
+    _cached_container_info: Optional[ContainerInspectInfo] = field(default=None, init=False, repr=False)
 
     def __post_init__(self) -> None:
         if self.Publishers:
@@ -159,6 +147,28 @@ def reload(self) -> None:
         # each time through get_container(), but we need this method for compatibility
         pass
 
+    def get_container_info(self) -> Optional[ContainerInspectInfo]:
+        """Get container information via docker inspect (lazy loaded).
+
+        Returns:
+            Container inspect information or None if container is not started.
+        """
+        if self._cached_container_info is not None:
+            return self._cached_container_info
+
+        if not self._docker_compose or not self.ID:
+            return None
+
+        try:
+            docker_client = self._docker_compose._get_docker_client()
+            self._cached_container_info = docker_client.get_container_inspect_info(self.ID)
+
+        except Exception as e:
+            logger.warning(f"Failed to get container info for {self.ID}: {e}")
+            self._cached_container_info = None
+
+        return self._cached_container_info
+
     @property
     def status(self) -> str:
         """Get container status for compatibility with wait strategies."""
@@ -233,6 +243,7 @@ class DockerCompose:
     quiet_pull: bool = False
     quiet_build: bool = False
     _wait_strategies: Optional[dict[str, Any]] = field(default=None, init=False, repr=False)
+    _docker_client: Optional[DockerClient] = field(default=None, init=False, repr=False)
 
     def __post_init__(self) -> None:
         if isinstance(self.compose_file_name, str):
@@ -597,3 +608,9 @@ def wait_for(self, url: str) -> "DockerCompose":
         with urlopen(url) as response:
             response.read()
         return self
+
+    def _get_docker_client(self) -> DockerClient:
+        """Get Docker client instance."""
+        if self._docker_client is None:
+            self._docker_client = DockerClient()
+        return self._docker_client

core/testcontainers/core/container.py

@@ -19,6 +19,7 @@
 from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.docker_client import DockerClient
 from testcontainers.core.exceptions import ContainerConnectException, ContainerStartException
+from testcontainers.core.inspect import ContainerInspectInfo
 from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID
 from testcontainers.core.network import Network
 from testcontainers.core.transferable import Transferable, TransferSpec, build_transfer_tar
@@ -104,6 +105,7 @@ def __init__(
 
         self._kwargs = kwargs
         self._wait_strategy: Optional[WaitStrategy] = _wait_strategy
+        self._cached_container_info: Optional[ContainerInspectInfo] = None
 
         self._transferable_specs: list[TransferSpec] = []
         if transferables:
@@ -328,6 +330,27 @@ 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).
+
+        Returns:
+            Container inspect information or None if container is not started.
+        """
+        if self._cached_container_info is not None:
+            return self._cached_container_info
+
+        if not self._container:
+            return None
+
+        try:
+            self._cached_container_info = self.get_docker_client().get_container_inspect_info(self._container.id)
+
+        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

core/testcontainers/core/docker_client.py

@@ -30,6 +30,7 @@
 from testcontainers.core.auth import DockerAuthInfo, parse_docker_auth_config
 from testcontainers.core.config import ConnectionMode
 from testcontainers.core.config import testcontainers_config as c
+from testcontainers.core.inspect import ContainerInspectInfo
 from testcontainers.core.labels import SESSION_ID, create_labels
 
 if TYPE_CHECKING:
@@ -275,6 +276,11 @@ def client_networks_create(self, name: str, param: dict[str, Any]) -> "DockerNet
         labels = create_labels("", param.get("labels"))
         return self.client.networks.create(name, **{**param, "labels": labels})
 
+    def get_container_inspect_info(self, container_id: str) -> "ContainerInspectInfo":
+        """Get container inspect information with fresh data."""
+        container = self.client.containers.get(container_id)
+        return ContainerInspectInfo.from_dict(container.attrs)
+
 
 def get_docker_host() -> Optional[str]:
     host = c.tc_properties_get_tc_host() or os.getenv("DOCKER_HOST")