-
Notifications
You must be signed in to change notification settings - Fork 370
feat(compose): add structured container inspect information #897
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
69bd44b
24f9fed
4b140a3
dc7960b
6f5f6bc
c30ca92
36b4488
1bd0054
a66bee2
d47b8d8
3622fc2
a9e92af
fa57d23
f69464a
527d5a5
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. do we need to care about invalidating?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is something I was thinking about, and I guess the answer depends, of course, network conditions, container status, and statistics can change along the way. But in a test scenario, you would usually create the container, run your tests, and then destroy it. I see that this new functionality would be used to collect the status at the end of the test, so this simple cache helps avoid extra overhead during test execution. What do you think? Otherwise, we have a few options here. The easiest one would be to always execute the docker inspect command and return fresh information, or we could define a default TTL for the data so that after this period, the next call retrieves updated data or something like this. |
||
|
|
||
| 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 | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok so i wrote this to implement this thing: https://fasterxml.github.io/jackson-annotations/javadoc/2.6/com/fasterxml/jackson/annotation/JsonIgnoreProperties.html#ignoreUnknown()
how do you do this in python? can we get rid of it? if yes, lets, if no: lets put it in a good final shareable place.
this is not that big of a deal, i will approve either way