19
19
import urllib
20
20
import urllib .parse
21
21
from collections .abc import Iterable
22
- from typing import Callable , Optional , TypeVar , Union
22
+ from typing import Any , Callable , Optional , TypeVar , Union , cast
23
23
24
24
import docker
25
25
from docker .models .containers import Container , ContainerCollection
@@ -59,7 +59,7 @@ class DockerClient:
59
59
Thin wrapper around :class:`docker.DockerClient` for a more functional interface.
60
60
"""
61
61
62
- def __init__ (self , ** kwargs ) -> None :
62
+ def __init__ (self , ** kwargs : Any ) -> None :
63
63
docker_host = get_docker_host ()
64
64
65
65
if docker_host :
@@ -82,14 +82,14 @@ def run(
82
82
self ,
83
83
image : str ,
84
84
command : Optional [Union [str , list [str ]]] = None ,
85
- environment : Optional [dict ] = None ,
86
- ports : Optional [dict ] = None ,
85
+ environment : Optional [dict [ str , str ] ] = None ,
86
+ ports : Optional [dict [ int , Optional [ int ]] ] = None ,
87
87
labels : Optional [dict [str , str ]] = None ,
88
88
detach : bool = False ,
89
89
stdout : bool = True ,
90
90
stderr : bool = False ,
91
91
remove : bool = False ,
92
- ** kwargs ,
92
+ ** kwargs : Any ,
93
93
) -> Container :
94
94
# If the user has specified a network, we'll assume the user knows best
95
95
if "network" not in kwargs and not get_docker_host ():
@@ -112,7 +112,7 @@ def run(
112
112
return container
113
113
114
114
@_wrapped_image_collection
115
- def build (self , path : str , tag : str , rm : bool = True , ** kwargs ) -> tuple [Image , Iterable [dict ]]:
115
+ def build (self , path : str , tag : str , rm : bool = True , ** kwargs : Any ) -> tuple [Image , Iterable [dict [ str , Any ] ]]:
116
116
"""
117
117
Build a Docker image from a directory containing the Dockerfile.
118
118
@@ -151,43 +151,43 @@ def find_host_network(self) -> Optional[str]:
151
151
except ipaddress .AddressValueError :
152
152
continue
153
153
if docker_host in subnet :
154
- return network .name
154
+ return cast ( str , network .name )
155
155
except (ipaddress .AddressValueError , OSError ):
156
156
pass
157
157
return None
158
158
159
- def port (self , container_id : str , port : int ) -> int :
159
+ def port (self , container_id : str , port : int ) -> str :
160
160
"""
161
161
Lookup the public-facing port that is NAT-ed to :code:`port`.
162
162
"""
163
163
port_mappings = self .client .api .port (container_id , port )
164
164
if not port_mappings :
165
165
raise ConnectionError (f"Port mapping for container { container_id } and port { port } is " "not available" )
166
- return port_mappings [0 ]["HostPort" ]
166
+ return cast ( str , port_mappings [0 ]["HostPort" ])
167
167
168
- def get_container (self , container_id : str ) -> Container :
168
+ def get_container (self , container_id : str ) -> dict [ str , Any ] :
169
169
"""
170
170
Get the container with a given identifier.
171
171
"""
172
172
containers = self .client .api .containers (filters = {"id" : container_id })
173
173
if not containers :
174
174
raise RuntimeError (f"Could not get container with id { container_id } " )
175
- return containers [0 ]
175
+ return cast ( dict [ str , Any ], containers [0 ])
176
176
177
177
def bridge_ip (self , container_id : str ) -> str :
178
178
"""
179
179
Get the bridge ip address for a container.
180
180
"""
181
181
container = self .get_container (container_id )
182
182
network_name = self .network_name (container_id )
183
- return container ["NetworkSettings" ]["Networks" ][network_name ]["IPAddress" ]
183
+ return str ( container ["NetworkSettings" ]["Networks" ][network_name ]["IPAddress" ])
184
184
185
185
def network_name (self , container_id : str ) -> str :
186
186
"""
187
187
Get the name of the network this container runs on
188
188
"""
189
189
container = self .get_container (container_id )
190
- name = container ["HostConfig" ]["NetworkMode" ]
190
+ name = str ( container ["HostConfig" ]["NetworkMode" ])
191
191
if name == "default" :
192
192
return "bridge"
193
193
return name
@@ -198,7 +198,7 @@ def gateway_ip(self, container_id: str) -> str:
198
198
"""
199
199
container = self .get_container (container_id )
200
200
network_name = self .network_name (container_id )
201
- return container ["NetworkSettings" ]["Networks" ][network_name ]["Gateway" ]
201
+ return str ( container ["NetworkSettings" ]["Networks" ][network_name ]["Gateway" ])
202
202
203
203
def get_connection_mode (self ) -> ConnectionMode :
204
204
"""
@@ -235,9 +235,10 @@ def host(self) -> str:
235
235
return "localhost"
236
236
if "http" in url .scheme or "tcp" in url .scheme and url .hostname :
237
237
# see https://github.com/testcontainers/testcontainers-python/issues/415
238
- if url .hostname == "localnpipe" and utils .is_windows ():
238
+ hostname = url .hostname
239
+ if not hostname or (hostname == "localnpipe" and utils .is_windows ()):
239
240
return "localhost"
240
- return url .hostname
241
+ return cast ( str , url .hostname )
241
242
if utils .inside_container () and ("unix" in url .scheme or "npipe" in url .scheme ):
242
243
ip_address = utils .default_gateway_ip ()
243
244
if ip_address :
@@ -251,9 +252,9 @@ def login(self, auth_config: DockerAuthInfo) -> None:
251
252
login_info = self .client .login (** auth_config ._asdict ())
252
253
LOGGER .debug (f"logged in using { login_info } " )
253
254
254
- def client_networks_create (self , name : str , param : dict ) :
255
+ def client_networks_create (self , name : str , param : dict [ str , Any ]) -> dict [ str , Any ] :
255
256
labels = create_labels ("" , param .get ("labels" ))
256
- return self .client .networks .create (name , ** {** param , "labels" : labels })
257
+ return cast ( dict [ str , Any ], self .client .networks .create (name , ** {** param , "labels" : labels }) )
257
258
258
259
259
260
def get_docker_host () -> Optional [str ]:
0 commit comments