Remove deprecated mypy option

podman/api/http_utils.py

@@ -7,7 +7,9 @@
 from collections.abc import Mapping
 
 
-def prepare_filters(filters: Union[str, list[str], Mapping[str, str]]) -> Optional[str]:
+def prepare_filters(
+    filters: Union[str, list[str], Mapping[str, Any], None],
+) -> Optional[str]:
     """Return filters as an URL quoted JSON dict[str, list[Any]]."""
 
     if filters is None or len(filters) == 0:

podman/client.py

@@ -127,11 +127,10 @@ def from_env(
         environment = environment or os.environ
         credstore_env = credstore_env or {}
 
-        if version == "auto":
-            version = None
+        api_version: Optional[str] = None if version == "auto" else version
 
         kwargs = {
-            'version': version,
+            'version': api_version,
             'timeout': timeout,
             'tls': False,
             'credstore_env': credstore_env,

podman/domain/containers.py

@@ -116,7 +116,7 @@ def commit(self, repository: str = None, tag: str = None, **kwargs) -> Image:
             "repo": repository,
             "tag": tag,
         }
-        response = self.client.post("/commit", params=params)
+        response = self.api.post("/commit", params=params)
         response.raise_for_status()
 
         body = response.json()
@@ -128,7 +128,7 @@ def diff(self) -> list[dict[str, int]]:
         Raises:
             APIError: when service reports an error
         """
-        response = self.client.get(f"/containers/{self.id}/changes")
+        response = self.api.get(f"/containers/{self.id}/changes")
         response.raise_for_status()
         return response.json()
 
@@ -207,11 +207,11 @@ def exec_run(
         stream = stream and not detach
 
         # create the exec instance
-        response = self.client.post(f"/containers/{self.name}/exec", data=json.dumps(data))
+        response = self.api.post(f"/containers/{self.name}/exec", data=json.dumps(data))
         response.raise_for_status()
         exec_id = response.json()['Id']
         # start the exec instance, this will store command output
-        start_resp = self.client.post(
+        start_resp = self.api.post(
             f"/exec/{exec_id}/start", data=json.dumps({"Detach": detach, "Tty": tty}), stream=stream
         )
         start_resp.raise_for_status()
@@ -220,7 +220,7 @@ def exec_run(
             return None, api.stream_frames(start_resp, demux=demux)
 
         # get and return exec information
-        response = self.client.get(f"/exec/{exec_id}/json")
+        response = self.api.get(f"/exec/{exec_id}/json")
         response.raise_for_status()
         if demux:
             stdout_data, stderr_data = demux_output(start_resp.content)
@@ -240,7 +240,7 @@ def export(self, chunk_size: int = api.DEFAULT_CHUNK_SIZE) -> Iterator[bytes]:
             NotFound: when container has been removed from service
             APIError: when service reports an error
         """
-        response = self.client.get(f"/containers/{self.id}/export", stream=True)
+        response = self.api.get(f"/containers/{self.id}/export", stream=True)
         response.raise_for_status()
 
         yield from response.iter_content(chunk_size=chunk_size)
@@ -258,7 +258,7 @@ def get_archive(
             First item is a raw tar data stream.
             Second item is a dict containing os.stat() information on the specified path.
         """
-        response = self.client.get(f"/containers/{self.id}/archive", params={"path": [path]})
+        response = self.api.get(f"/containers/{self.id}/archive", params={"path": [path]})
         response.raise_for_status()
 
         stat = response.headers.get("x-docker-container-path-stat", None)
@@ -267,7 +267,7 @@ def get_archive(
 
     def init(self) -> None:
         """Initialize the container."""
-        response = self.client.post(f"/containers/{self.id}/init")
+        response = self.api.post(f"/containers/{self.id}/init")
         response.raise_for_status()
 
     def inspect(self) -> dict:
@@ -276,7 +276,7 @@ def inspect(self) -> dict:
         Raises:
             APIError: when service reports an error
         """
-        response = self.client.get(f"/containers/{self.id}/json")
+        response = self.api.get(f"/containers/{self.id}/json")
         response.raise_for_status()
         return response.json()
 
@@ -286,7 +286,7 @@ def kill(self, signal: Union[str, int, None] = None) -> None:
         Raises:
             APIError: when service reports an error
         """
-        response = self.client.post(f"/containers/{self.id}/kill", params={"signal": signal})
+        response = self.api.post(f"/containers/{self.id}/kill", params={"signal": signal})
         response.raise_for_status()
 
     def logs(self, **kwargs) -> Union[bytes, Iterator[bytes]]:
@@ -317,7 +317,7 @@ def logs(self, **kwargs) -> Union[bytes, Iterator[bytes]]:
             "until": api.prepare_timestamp(kwargs.get("until")),
         }
 
-        response = self.client.get(f"/containers/{self.id}/logs", stream=stream, params=params)
+        response = self.api.get(f"/containers/{self.id}/logs", stream=stream, params=params)
         response.raise_for_status()
 
         if stream:
@@ -326,7 +326,7 @@ def logs(self, **kwargs) -> Union[bytes, Iterator[bytes]]:
 
     def pause(self) -> None:
         """Pause processes within the container."""
-        response = self.client.post(f"/containers/{self.id}/pause")
+        response = self.api.post(f"/containers/{self.id}/pause")
         response.raise_for_status()
 
     def put_archive(self, path: str, data: bytes = None) -> bool:
@@ -349,9 +349,7 @@ def put_archive(self, path: str, data: bytes = None) -> bool:
         if data is None:
             data = api.create_tar("/", path)
 
-        response = self.client.put(
-            f"/containers/{self.id}/archive", params={"path": path}, data=data
-        )
+        response = self.api.put(f"/containers/{self.id}/archive", params={"path": path}, data=data)
         return response.ok
 
     def remove(self, **kwargs) -> None:
@@ -375,7 +373,7 @@ def rename(self, name: str) -> None:
         if not name:
             raise ValueError("'name' is a required argument.")
 
-        response = self.client.post(f"/containers/{self.id}/rename", params={"name": name})
+        response = self.api.post(f"/containers/{self.id}/rename", params={"name": name})
         response.raise_for_status()
 
         self.attrs["Name"] = name  # shortcut to avoid needing reload()
@@ -391,7 +389,7 @@ def resize(self, height: int = None, width: int = None) -> None:
             "h": height,
             "w": width,
         }
-        response = self.client.post(f"/containers/{self.id}/resize", params=params)
+        response = self.api.post(f"/containers/{self.id}/resize", params=params)
         response.raise_for_status()
 
     def restart(self, **kwargs) -> None:
@@ -405,7 +403,7 @@ def restart(self, **kwargs) -> None:
         if kwargs.get("timeout"):
             post_kwargs["timeout"] = float(params["timeout"]) * 1.5
 
-        response = self.client.post(f"/containers/{self.id}/restart", params=params, **post_kwargs)
+        response = self.api.post(f"/containers/{self.id}/restart", params=params, **post_kwargs)
         response.raise_for_status()
 
     def start(self, **kwargs) -> None:
@@ -414,7 +412,7 @@ def start(self, **kwargs) -> None:
         Keyword Args:
             detach_keys: Override the key sequence for detaching a container (Podman only)
         """
-        response = self.client.post(
+        response = self.api.post(
             f"/containers/{self.id}/start", params={"detachKeys": kwargs.get("detach_keys")}
         )
         response.raise_for_status()
@@ -441,7 +439,7 @@ def stats(
             "stream": stream,
         }
 
-        response = self.client.get("/containers/stats", params=params, stream=stream)
+        response = self.api.get("/containers/stats", params=params, stream=stream)
         response.raise_for_status()
 
         if stream:
@@ -463,7 +461,7 @@ def stop(self, **kwargs) -> None:
         if kwargs.get("timeout"):
             post_kwargs["timeout"] = float(params["timeout"]) * 1.5
 
-        response = self.client.post(f"/containers/{self.id}/stop", params=params, **post_kwargs)
+        response = self.api.post(f"/containers/{self.id}/stop", params=params, **post_kwargs)
         response.raise_for_status()
 
         if response.status_code == requests.codes.no_content:
@@ -493,7 +491,7 @@ def top(self, **kwargs) -> Union[Iterator[dict[str, Any]], dict[str, Any]]:
             "stream": stream,
             "ps_args": kwargs.get("ps_args"),
         }
-        response = self.client.get(f"/containers/{self.id}/top", params=params, stream=stream)
+        response = self.api.get(f"/containers/{self.id}/top", params=params, stream=stream)
         response.raise_for_status()
 
         if stream:
@@ -503,7 +501,7 @@ def top(self, **kwargs) -> Union[Iterator[dict[str, Any]], dict[str, Any]]:
 
     def unpause(self) -> None:
         """Unpause processes in container."""
-        response = self.client.post(f"/containers/{self.id}/unpause")
+        response = self.api.post(f"/containers/{self.id}/unpause")
         response.raise_for_status()
 
     def update(self, **kwargs) -> None:
@@ -724,7 +722,7 @@ def update(self, **kwargs) -> None:
         if kwargs.get("restart_retries"):
             params["restartRetries"] = kwargs.get("restart_retries")
 
-        response = self.client.post(
+        response = self.api.post(
             f"/containers/{self.id}/update", params=params, data=json.dumps(data)
         )
         response.raise_for_status()
@@ -764,6 +762,6 @@ def wait(self, **kwargs) -> int:
         # This API endpoint responds with a JSON encoded integer, the exit code of the container.
         # See:
         # https://docs.podman.io/en/latest/_static/api.html#tag/containers/operation/ContainerWaitLibpod
-        response = self.client.post(f"/containers/{self.id}/wait", params=params, timeout=timeout)
+        response = self.api.post(f"/containers/{self.id}/wait", params=params, timeout=timeout)
         response.raise_for_status()
         return response.json()

podman/domain/containers_create.py

@@ -421,7 +421,7 @@ def create(
         payload = self._render_payload(payload)
         payload = api.prepare_body(payload)
 
-        response = self.client.post(
+        response = self.api.post(
             "/containers/create",
             headers={"content-type": "application/json"},
             data=payload,

podman/domain/containers_manager.py

@@ -3,7 +3,7 @@
 import logging
 import urllib
 from collections.abc import Mapping
-from typing import Any, Union
+from typing import Any, Optional, Union
 
 from podman import api
 from podman.domain.containers import Container
@@ -24,7 +24,7 @@ def resource(self):
         return Container
 
     def exists(self, key: str) -> bool:
-        response = self.client.get(f"/containers/{key}/exists")
+        response = self.api.get(f"/containers/{key}/exists")
         return response.ok
 
     def get(self, key: str, **kwargs) -> Container:
@@ -46,7 +46,7 @@ def get(self, key: str, **kwargs) -> Container:
         compatible = kwargs.get("compatible", False)
 
         container_id = urllib.parse.quote_plus(key)
-        response = self.client.get(f"/containers/{container_id}/json", compatible=compatible)
+        response = self.api.get(f"/containers/{container_id}/json", compatible=compatible)
         response.raise_for_status()
         return self.prepare_model(attrs=response.json())
 
@@ -105,7 +105,7 @@ def list(self, **kwargs) -> list[Container]:
         # filters formatted last because some kwargs may need to be mapped into filters
         params["filters"] = api.prepare_filters(params["filters"])
 
-        response = self.client.get("/containers/json", params=params, compatible=compatible)
+        response = self.api.get("/containers/json", params=params, compatible=compatible)
         response.raise_for_status()
 
         containers: list[Container] = [self.prepare_model(attrs=i) for i in response.json()]
@@ -121,7 +121,7 @@ def list(self, **kwargs) -> list[Container]:
 
         return containers
 
-    def prune(self, filters: Mapping[str, str] = None) -> dict[str, Any]:
+    def prune(self, filters: Optional[Mapping[str, str]] = None) -> dict[str, Any]:
         """Delete stopped containers.
 
         Args:
@@ -138,7 +138,7 @@ def prune(self, filters: Mapping[str, str] = None) -> dict[str, Any]:
             APIError: when service reports an error
         """
         params = {"filters": api.prepare_filters(filters)}
-        response = self.client.post("/containers/prune", params=params)
+        response = self.api.post("/containers/prune", params=params)
         response.raise_for_status()
 
         results = {"ContainersDeleted": [], "SpaceReclaimed": 0}
@@ -173,5 +173,5 @@ def remove(self, container_id: Union[Container, str], **kwargs):
         # v is used for the compat endpoint while volumes is used for the libpod endpoint
         params = {"v": kwargs.get("v"), "force": kwargs.get("force"), "volumes": kwargs.get("v")}
 
-        response = self.client.delete(f"/containers/{container_id}", params=params)
+        response = self.api.delete(f"/containers/{container_id}", params=params)
         response.raise_for_status()

podman/domain/containers_run.py

@@ -108,6 +108,6 @@ def remove_container(container_object: Container) -> None:
             container.remove()
 
         if exit_status != 0:
-            raise ContainerError(container, exit_status, command, image_id, log_iter)
+            raise ContainerError(container, exit_status, command or [], image_id, log_iter)
 
         return log_iter if kwargs.get("stream", False) or log_iter is None else b"".join(log_iter)  # type: ignore[return-value]

podman/domain/images.py

@@ -49,7 +49,7 @@ def history(self) -> list[dict[str, Any]]:
             APIError: when service returns an error
         """
 
-        response = self.client.get(f"/images/{self.id}/history")
+        response = self.api.get(f"/images/{self.id}/history")
         response.raise_for_status(not_found=ImageNotFound)
         return response.json()
 
@@ -105,7 +105,7 @@ def save(
                     raise InvalidArgument(f"'{named}' is not a valid tag for this image")
                 img = urllib.parse.quote(named)
 
-        response = self.client.get(
+        response = self.api.get(
             f"/images/{img}/get", params={"format": ["docker-archive"]}, stream=True
         )
         response.raise_for_status(not_found=ImageNotFound)
@@ -132,7 +132,7 @@ def tag(
             APIError: when service returns an error
         """
         params = {"repo": repository, "tag": tag}
-        response = self.client.post(f"/images/{self.id}/tag", params=params)
+        response = self.api.post(f"/images/{self.id}/tag", params=params)
         if response.ok:
             return True
 

podman/domain/images_build.py

@@ -123,7 +123,7 @@ def build(self, **kwargs) -> tuple[Image, Iterator[bytes]]:
         if kwargs.get("timeout"):
             post_kwargs["timeout"] = float(kwargs.get("timeout"))
 
-        response = self.client.post(  # type: ignore[attr-defined]
+        response = self.api.post(  # type: ignore[attr-defined]
             "/build",
             params=params,
             data=body,