Bases: Block
Block used to manage settings for interacting with a Docker host.
Attributes:
Name |
Type |
Description |
base_url |
Optional[str]
|
URL to the Docker server, e.g. unix:///var/run/docker.sock
or tcp://127.0.0.1:1234 . If this is not set, the client will
be configured from environment variables.
|
version |
str
|
The version of the API to use. Set to auto to
automatically detect the server's version.
|
timeout |
Optional[int]
|
Default timeout for API calls, in seconds.
|
max_pool_size |
Optional[int]
|
The maximum number of connections to save in the pool.
|
client_kwargs |
Dict[str, Any]
|
Additional keyword arguments to pass to
docker.from_env() or DockerClient .
|
Examples:
Get a Docker Host client.
from prefect_docker import DockerHost
docker_host = DockerHost(
base_url="tcp://127.0.0.1:1234",
max_pool_size=4
)
with docker_host.get_client() as client:
... # Use the client for Docker operations
Source code in prefect_docker/host.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120 | class DockerHost(Block):
"""
Block used to manage settings for interacting with a Docker host.
Attributes:
base_url: URL to the Docker server, e.g. `unix:///var/run/docker.sock`
or `tcp://127.0.0.1:1234`. If this is not set, the client will
be configured from environment variables.
version: The version of the API to use. Set to auto to
automatically detect the server's version.
timeout: Default timeout for API calls, in seconds.
max_pool_size: The maximum number of connections to save in the pool.
client_kwargs: Additional keyword arguments to pass to
`docker.from_env()` or `DockerClient`.
Examples:
Get a Docker Host client.
```python
from prefect_docker import DockerHost
docker_host = DockerHost(
base_url="tcp://127.0.0.1:1234",
max_pool_size=4
)
with docker_host.get_client() as client:
... # Use the client for Docker operations
```
"""
_block_type_name = "Docker Host"
_logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/14a315b79990200db7341e42553e23650b34bb96-250x250.png" # noqa
_description = "Store settings for interacting with a Docker host."
base_url: Optional[str] = Field(
default=None,
description="URL to the Docker host.",
title="Base URL",
example="unix:///var/run/docker.sock",
)
version: str = Field(default="auto", description="The version of the API to use")
timeout: Optional[int] = Field(
default=None, description="Default timeout for API calls, in seconds."
)
max_pool_size: Optional[int] = Field(
default=None,
description="The maximum number of connections to save in the pool.",
)
client_kwargs: Dict[str, Any] = Field(
default_factory=dict,
title="Additional Configuration",
description=(
"Additional keyword arguments to pass to "
"`docker.from_env()` or `DockerClient`."
),
)
def get_client(self) -> docker.DockerClient:
"""
Gets a Docker Client to communicate with a Docker host.
Returns:
A Docker Client.
"""
logger = get_run_logger()
client_kwargs = {
"version": self.version,
"timeout": self.timeout,
"max_pool_size": self.max_pool_size,
**self.client_kwargs,
}
client_kwargs = {
key: value for key, value in client_kwargs.items() if value is not None
}
if self.base_url is None:
logger.debug(
f"Creating a Docker client from "
f"environment variables, using {self.version} version."
)
client = _ContextManageableDockerClient.from_env(**client_kwargs)
else:
logger.debug(
f"Creating a Docker client to {self.base_url} "
f"using {self.version} version."
)
client = _ContextManageableDockerClient(
base_url=self.base_url, **client_kwargs
)
return client
|
Functions
get_client
Gets a Docker Client to communicate with a Docker host.
Returns:
Type |
Description |
DockerClient
|
|
Source code in prefect_docker/host.py
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120 | def get_client(self) -> docker.DockerClient:
"""
Gets a Docker Client to communicate with a Docker host.
Returns:
A Docker Client.
"""
logger = get_run_logger()
client_kwargs = {
"version": self.version,
"timeout": self.timeout,
"max_pool_size": self.max_pool_size,
**self.client_kwargs,
}
client_kwargs = {
key: value for key, value in client_kwargs.items() if value is not None
}
if self.base_url is None:
logger.debug(
f"Creating a Docker client from "
f"environment variables, using {self.version} version."
)
client = _ContextManageableDockerClient.from_env(**client_kwargs)
else:
logger.debug(
f"Creating a Docker client to {self.base_url} "
f"using {self.version} version."
)
client = _ContextManageableDockerClient(
base_url=self.base_url, **client_kwargs
)
return client
|