From 6c466780822e4bb450405eea0aa5c71c894dce8e Mon Sep 17 00:00:00 2001
From: Povilas Kanapickas <povilas@radix.lt>
Date: Sat, 24 May 2025 17:10:16 +0300
Subject: [PATCH] Fix mypy warnings

Signed-off-by: Povilas Kanapickas <povilas@radix.lt>
---
 podman_compose.py | 519 +++++++++++++++++++++++++++-------------------
 1 file changed, 300 insertions(+), 219 deletions(-)

diff --git a/podman_compose.py b/podman_compose.py
index 4fe044e..05bdcd1 100755
--- a/podman_compose.py
+++ b/podman_compose.py
@@ -28,6 +28,11 @@ import tempfile
 import urllib.parse
 from asyncio import Task
 from enum import Enum
+from typing import Any
+from typing import Callable
+from typing import Iterable
+from typing import Union
+from typing import overload
 
 # import fnmatch
 # fnmatch.fnmatchcase(env, "*_HOST")
@@ -41,7 +46,7 @@ script = os.path.realpath(sys.argv[0])
 # helper functions
 
 
-def is_list(list_object):
+def is_list(list_object: Any) -> bool:
     return (
         not isinstance(list_object, str)
         and not isinstance(list_object, dict)
@@ -50,11 +55,17 @@ def is_list(list_object):
 
 
 # identity filter
-def filteri(a):
-    return filter(lambda i: i, a)
+def filteri(a: list[str]) -> list[str]:
+    return list(filter(lambda i: i, a))
 
 
-def try_int(i, fallback=None):
+@overload
+def try_int(i: int | str, fallback: int) -> int: ...
+@overload
+def try_int(i: int | str, fallback: None) -> int | None: ...
+
+
+def try_int(i: int | str, fallback: int | None = None) -> int | None:
     try:
         return int(i)
     except ValueError:
@@ -64,7 +75,7 @@ def try_int(i, fallback=None):
     return fallback
 
 
-def try_float(i, fallback=None):
+def try_float(i: int | str, fallback: float | None = None) -> float | None:
     try:
         return float(i)
     except ValueError:
@@ -100,7 +111,7 @@ t_re = re.compile(r"^(?:(\d+)[m:])?(?:(\d+(?:\.\d+)?)s?)?$")
 STOP_GRACE_PERIOD = "10"
 
 
-def str_to_seconds(txt):
+def str_to_seconds(txt: int | str | None) -> int | None:
     if not txt:
         return None
     if isinstance(txt, (int, float)):
@@ -117,19 +128,19 @@ def str_to_seconds(txt):
     return int(mins * 60.0 + sec)
 
 
-def ver_as_list(a):
+def ver_as_list(a: str) -> list[int]:
     return [try_int(i, i) for i in num_split_re.findall(a)]
 
 
-def strverscmp_lt(a, b):
+def strverscmp_lt(a: str, b: str) -> bool:
     a_ls = ver_as_list(a or "")
     b_ls = ver_as_list(b or "")
     return a_ls < b_ls
 
 
-def parse_short_mount(mount_str, basedir):
+def parse_short_mount(mount_str: str, basedir: str) -> dict[str, Any]:
     mount_a = mount_str.split(":")
-    mount_opt_dict = {}
+    mount_opt_dict: dict[str, Any] = {}
     mount_opt = None
     if len(mount_a) == 1:
         # Anonymous: Just specify a path and let the engine creates the volume
@@ -189,7 +200,9 @@ def parse_short_mount(mount_str, basedir):
 # unless it's anonymous-volume
 
 
-def fix_mount_dict(compose, mount_dict, srv_name):
+def fix_mount_dict(
+    compose: PodmanCompose, mount_dict: dict[str, Any], srv_name: str
+) -> dict[str, Any]:
     """
     in-place fix mount dictionary to:
     - define _vol to be the corresponding top-level volume
@@ -197,12 +210,14 @@ def fix_mount_dict(compose, mount_dict, srv_name):
     - if no source it would be generated
     """
     # if already applied nothing todo
+    assert compose.project_name is not None
+
     if "_vol" in mount_dict:
         return mount_dict
     if mount_dict["type"] == "volume":
         vols = compose.vols
         source = mount_dict.get("source")
-        vol = (vols.get(source, {}) or {}) if source else {}
+        vol = (vols.get(source, {}) or {}) if source else {}  # type: ignore[union-attr]
         name = vol.get("name")
         mount_dict["_vol"] = vol
         # handle anonymous or implied volume
@@ -254,7 +269,15 @@ var_re = re.compile(
 )
 
 
-def rec_subs(value, subs_dict):
+@overload
+def rec_subs(value: dict, subs_dict: dict[str, Any]) -> dict: ...
+@overload
+def rec_subs(value: str, subs_dict: dict[str, Any]) -> str: ...
+@overload
+def rec_subs(value: Iterable, subs_dict: dict[str, Any]) -> Iterable: ...
+
+
+def rec_subs(value: dict | str | Iterable, subs_dict: dict[str, Any]) -> dict | str | Iterable:
     """
     do bash-like substitution in value and if list of dictionary do that recursively
     """
@@ -271,7 +294,7 @@ def rec_subs(value, subs_dict):
         value = {k: rec_subs(v, subs_dict) for k, v in value.items()}
     elif isinstance(value, str):
 
-        def convert(m):
+        def convert(m: re.Match) -> str:
             if m.group("escaped") is not None:
                 return "$"
             name = m.group("named") or m.group("braced")
@@ -290,12 +313,12 @@ def rec_subs(value, subs_dict):
     return value
 
 
-def norm_as_list(src):
+def norm_as_list(src: dict[str, Any] | list[Any] | None) -> list[Any]:
     """
     given a dictionary {key1:value1, key2: None} or list
     return a list of ["key1=value1", "key2"]
     """
-    dst: list[str]
+    dst: list[Any]
     if src is None:
         dst = []
     elif isinstance(src, dict):
@@ -307,7 +330,7 @@ def norm_as_list(src):
     return dst
 
 
-def norm_as_dict(src):
+def norm_as_dict(src: None | dict[str, str | None] | list[str] | str) -> dict[str, str | None]:
     """
     given a list ["key1=value1", "key2"]
     return a dictionary {key1:value1, key2: None}
@@ -317,8 +340,8 @@ def norm_as_dict(src):
     elif isinstance(src, dict):
         dst = dict(src)
     elif is_list(src):
-        dst = [i.split("=", 1) for i in src if i]
-        dst = [(a if len(a) == 2 else (a[0], None)) for a in dst]
+        dst = [i.split("=", 1) for i in src if i]  # type: ignore[assignment]
+        dst = [(a if len(a) == 2 else (a[0], None)) for a in dst]  # type: ignore[assignment]
         dst = dict(dst)
     elif isinstance(src, str):
         key, value = src.split("=", 1) if "=" in src else (src, None)
@@ -328,7 +351,7 @@ def norm_as_dict(src):
     return dst
 
 
-def norm_ulimit(inner_value):
+def norm_ulimit(inner_value: dict | list | int | str) -> str:
     if isinstance(inner_value, dict):
         if not inner_value.keys() & {"soft", "hard"}:
             raise ValueError("expected at least one soft or hard limit")
@@ -336,15 +359,17 @@ def norm_ulimit(inner_value):
         hard = inner_value.get("hard", inner_value.get("soft"))
         return f"{soft}:{hard}"
     if is_list(inner_value):
-        return norm_ulimit(norm_as_dict(inner_value))
+        return norm_ulimit(norm_as_dict(inner_value))  # type: ignore[arg-type]
     # if int or string return as is
-    return inner_value
+    return inner_value  # type: ignore[return-value]
 
 
-def default_network_name_for_project(compose, net, is_ext):
+def default_network_name_for_project(compose: PodmanCompose, net: str, is_ext: Any) -> str:
     if is_ext:
         return net
 
+    assert compose.project_name is not None
+
     default_net_name_compat = compose.x_podman.get("default_net_name_compat", False)
     if default_net_name_compat is True:
         return f"{compose.project_name.replace('-', '')}_{net}"
@@ -360,7 +385,9 @@ def default_network_name_for_project(compose, net, is_ext):
 #    return [pod], containers
 
 
-def transform(args, project_name, given_containers):
+def transform(
+    args: Any, project_name: str, given_containers: list[Any]
+) -> tuple[list[dict], list[dict]]:
     in_pod = str(args.in_pod).lower()
     pod_name = None
     pods = []
@@ -379,7 +406,7 @@ def transform(args, project_name, given_containers):
     return pods, containers
 
 
-async def assert_volume(compose, mount_dict):
+async def assert_volume(compose: PodmanCompose, mount_dict: dict[str, Any]) -> None:
     """
     inspect volume to get directory
     create volume if needed
@@ -429,8 +456,11 @@ async def assert_volume(compose, mount_dict):
         _ = (await compose.podman.output([], "volume", ["inspect", vol_name])).decode("utf-8")
 
 
-def mount_desc_to_mount_args(compose, mount_desc, srv_name, cnt_name):  # pylint: disable=unused-argument
-    mount_type = mount_desc.get("type")
+def mount_desc_to_mount_args(
+    compose: PodmanCompose, mount_desc: dict[str, Any], srv_name: str, cnt_name: str
+) -> str:  # pylint: disable=unused-argument
+    mount_type: str | None = mount_desc.get("type")
+    assert mount_type is not None
     vol = mount_desc.get("_vol") if mount_type == "volume" else None
     source = vol["name"] if vol else mount_desc.get("source")
     target = mount_desc["target"]
@@ -465,7 +495,7 @@ def mount_desc_to_mount_args(compose, mount_desc, srv_name, cnt_name):  # pylint
     raise ValueError("unknown mount type:" + mount_type)
 
 
-def ulimit_to_ulimit_args(ulimit, podman_args):
+def ulimit_to_ulimit_args(ulimit: str | dict[str, Any] | list[Any], podman_args: list[str]) -> None:
     if ulimit is not None:
         # ulimit can be a single value, i.e. ulimit: host
         if isinstance(ulimit, str):
@@ -474,25 +504,27 @@ def ulimit_to_ulimit_args(ulimit, podman_args):
         else:
             ulimit = norm_as_dict(ulimit)
             ulimit = [
-                "{}={}".format(ulimit_key, norm_ulimit(inner_value))
-                for ulimit_key, inner_value in ulimit.items()
+                "{}={}".format(ulimit_key, norm_ulimit(inner_value))  # type: ignore[arg-type]
+                for ulimit_key, inner_value in ulimit.items()  # type: ignore[union-attr]
             ]
             for i in ulimit:
                 podman_args.extend(["--ulimit", i])
 
 
-def container_to_ulimit_args(cnt, podman_args):
+def container_to_ulimit_args(cnt: dict[str, Any], podman_args: list[str]) -> None:
     ulimit_to_ulimit_args(cnt.get("ulimits", []), podman_args)
 
 
-def container_to_ulimit_build_args(cnt, podman_args):
+def container_to_ulimit_build_args(cnt: dict[str, Any], podman_args: list[str]) -> None:
     build = cnt.get("build")
 
     if build is not None:
         ulimit_to_ulimit_args(build.get("ulimits", []), podman_args)
 
 
-def mount_desc_to_volume_args(compose, mount_desc, srv_name, cnt_name):  # pylint: disable=unused-argument
+def mount_desc_to_volume_args(
+    compose: PodmanCompose, mount_desc: dict[str, Any], srv_name: str, cnt_name: str
+) -> str:  # pylint: disable=unused-argument
     mount_type = mount_desc["type"]
     if mount_type not in ("bind", "volume"):
         raise ValueError("unknown mount type:" + mount_type)
@@ -501,7 +533,7 @@ def mount_desc_to_volume_args(compose, mount_desc, srv_name, cnt_name):  # pylin
     if not source:
         raise ValueError(f"missing mount source for {mount_type} on {srv_name}")
     target = mount_desc["target"]
-    opts = []
+    opts: list[str] = []
 
     propagations = set(filteri(mount_desc.get(mount_type, {}).get("propagation", "").split(",")))
     if mount_type != "bind":
@@ -532,7 +564,9 @@ def mount_desc_to_volume_args(compose, mount_desc, srv_name, cnt_name):  # pylin
     return args
 
 
-def get_mnt_dict(compose, cnt, volume):
+def get_mnt_dict(
+    compose: PodmanCompose, cnt: dict[str, Any], volume: str | dict[str, Any]
+) -> dict[str, Any]:
     srv_name = cnt["_service"]
     basedir = compose.dirname
     if isinstance(volume, str):
@@ -540,7 +574,9 @@ def get_mnt_dict(compose, cnt, volume):
     return fix_mount_dict(compose, volume, srv_name)
 
 
-async def get_mount_args(compose, cnt, volume):
+async def get_mount_args(
+    compose: PodmanCompose, cnt: dict[str, Any], volume: str | dict[str, Any]
+) -> list[str]:
     volume = get_mnt_dict(compose, cnt, volume)
     srv_name = cnt["_service"]
     mount_type = volume["type"]
@@ -566,11 +602,18 @@ async def get_mount_args(compose, cnt, volume):
     return ["--mount", args]
 
 
-def get_secret_args(compose, cnt, secret, podman_is_building=False):
+def get_secret_args(
+    compose: PodmanCompose,
+    cnt: dict[str, Any],
+    secret: str | dict[str, Any],
+    podman_is_building: bool = False,
+) -> list[str]:
     """
     podman_is_building: True if we are preparing arguments for an invocation of "podman build"
                         False if we are preparing for something else like "podman run"
     """
+    assert compose.declared_secrets is not None
+
     secret_name = secret if isinstance(secret, str) else secret.get("source")
     if not secret_name or secret_name not in compose.declared_secrets.keys():
         raise ValueError(f'ERROR: undeclared secret: "{secret}", service: {cnt["_service"]}')
@@ -671,12 +714,12 @@ def get_secret_args(compose, cnt, secret, podman_is_building=False):
     )
 
 
-def container_to_res_args(cnt, podman_args):
+def container_to_res_args(cnt: dict[str, Any], podman_args: list[str]) -> None:
     container_to_cpu_res_args(cnt, podman_args)
     container_to_gpu_res_args(cnt, podman_args)
 
 
-def container_to_gpu_res_args(cnt, podman_args):
+def container_to_gpu_res_args(cnt: dict[str, Any], podman_args: list[str]) -> None:
     # https://docs.docker.com/compose/gpu-support/
     # https://docs.nvidia.com/datacenter/cloud-native/container-toolkit/latest/cdi-support.html
 
@@ -727,11 +770,11 @@ def container_to_gpu_res_args(cnt, podman_args):
         podman_args.append("--security-opt=label=disable")
 
 
-def container_to_cpu_res_args(cnt, podman_args):
+def container_to_cpu_res_args(cnt: dict[str, Any], podman_args: list[str]) -> None:
     # v2: https://docs.docker.com/compose/compose-file/compose-file-v2/#cpu-and-other-resources
     # cpus, cpu_shares, mem_limit, mem_reservation
-    cpus_limit_v2 = try_float(cnt.get("cpus"), None)
-    cpu_shares_v2 = try_int(cnt.get("cpu_shares"), None)
+    cpus_limit_v2 = try_float(cnt.get("cpus"), None)  # type: ignore[arg-type]
+    cpu_shares_v2 = try_int(cnt.get("cpu_shares"), None)  # type: ignore[arg-type]
     mem_limit_v2 = cnt.get("mem_limit")
     mem_res_v2 = cnt.get("mem_reservation")
     # v3: https://docs.docker.com/compose/compose-file/compose-file-v3/#resources
@@ -787,7 +830,7 @@ def container_to_cpu_res_args(cnt, podman_args):
         podman_args.extend(["--pids-limit", str(final_pids_limit)])
 
 
-def port_dict_to_str(port_desc):
+def port_dict_to_str(port_desc: dict[str, Any]) -> str:
     # NOTE: `mode: host|ingress` is ignored
     cnt_port = port_desc.get("target")
     published = port_desc.get("published", "")
@@ -804,11 +847,14 @@ def port_dict_to_str(port_desc):
     return ret
 
 
-def norm_ports(ports_in):
+def norm_ports(
+    ports_in: None | str | list[str | dict[str, Any] | int] | dict[str, Any] | int,
+) -> list[str]:
     if not ports_in:
         ports_in = []
     if isinstance(ports_in, str):
         ports_in = [ports_in]
+    assert isinstance(ports_in, list)
     ports_out = []
     for port in ports_in:
         if isinstance(port, dict):
@@ -821,7 +867,7 @@ def norm_ports(ports_in):
     return ports_out
 
 
-def get_network_create_args(net_desc, proj_name, net_name):
+def get_network_create_args(net_desc: dict[str, Any], proj_name: str, net_name: str) -> list[str]:
     args = [
         "create",
         "--label",
@@ -873,7 +919,7 @@ def get_network_create_args(net_desc, proj_name, net_name):
     return args
 
 
-async def assert_cnt_nets(compose, cnt):
+async def assert_cnt_nets(compose: PodmanCompose, cnt: dict[str, Any]) -> None:
     """
     create missing networks
     """
@@ -881,10 +927,12 @@ async def assert_cnt_nets(compose, cnt):
     if net:
         return
 
+    assert compose.project_name is not None
+
     cnt_nets = cnt.get("networks")
     if cnt_nets and isinstance(cnt_nets, dict):
         cnt_nets = list(cnt_nets.keys())
-    cnt_nets = norm_as_list(cnt_nets or compose.default_net)
+    cnt_nets = norm_as_list(cnt_nets or compose.default_net)  # type: ignore[arg-type]
     for net in cnt_nets:
         net_desc = compose.networks[net] or {}
         is_ext = net_desc.get("external")
@@ -901,9 +949,10 @@ async def assert_cnt_nets(compose, cnt):
             await compose.podman.output([], "network", ["exists", net_name])
 
 
-def get_net_args_from_network_mode(compose, cnt):
+def get_net_args_from_network_mode(compose: PodmanCompose, cnt: dict[str, Any]) -> list[str]:
     net_args = []
     net = cnt.get("network_mode")
+    assert isinstance(net, str)
     service_name = cnt["service_name"]
 
     if "networks" in cnt:
@@ -933,13 +982,13 @@ def get_net_args_from_network_mode(compose, cnt):
     elif net.startswith("bridge"):
         aliases_on_container = [service_name]
         if cnt.get("_aliases"):
-            aliases_on_container.extend(cnt.get("_aliases"))
+            aliases_on_container.extend(cnt.get("_aliases"))  # type: ignore[arg-type]
         net_options = [f"alias={alias}" for alias in aliases_on_container]
         mac_address = cnt.get("mac_address")
         if mac_address:
             net_options.append(f"mac={mac_address}")
 
-        net = f"{net}," if ":" in net else f"{net}:"
+        net = f"{net}," if ":" in net else f"{net}:"  # type: ignore[operator]
         net_args.append(f"--network={net}{','.join(net_options)}")
     else:
         log.fatal("unknown network_mode [%s]", net)
@@ -948,7 +997,7 @@ def get_net_args_from_network_mode(compose, cnt):
     return net_args
 
 
-def get_net_args(compose, cnt):
+def get_net_args(compose: PodmanCompose, cnt: dict[str, Any]) -> list[str]:
     net = cnt.get("network_mode")
     if net:
         return get_net_args_from_network_mode(compose, cnt)
@@ -956,7 +1005,7 @@ def get_net_args(compose, cnt):
     return get_net_args_from_networks(compose, cnt)
 
 
-def get_net_args_from_networks(compose, cnt):
+def get_net_args_from_networks(compose: PodmanCompose, cnt: dict[str, Any]) -> list[str]:
     net_args = []
     mac_address = cnt.get("mac_address")
     service_name = cnt["service_name"]
@@ -1000,8 +1049,8 @@ def get_net_args_from_networks(compose, cnt):
     for net_, net_config_ in multiple_nets.items():
         net_desc = compose.networks.get(net_) or {}
         is_ext = net_desc.get("external")
-        ext_desc = is_ext if isinstance(is_ext, str) else {}
-        default_net_name = default_network_name_for_project(compose, net_, is_ext)
+        ext_desc: dict[str, Any] = is_ext if isinstance(is_ext, str) else {}  # type: ignore[assignment]
+        default_net_name = default_network_name_for_project(compose, net_, is_ext)  # type: ignore[arg-type]
         net_name = ext_desc.get("name") or net_desc.get("name") or default_net_name
 
         interface_name = net_config_.get("x-podman.interface_name")
@@ -1045,7 +1094,9 @@ def get_net_args_from_networks(compose, cnt):
     return net_args
 
 
-async def container_to_args(compose, cnt, detached=True, no_deps=False):
+async def container_to_args(
+    compose: PodmanCompose, cnt: dict[str, Any], detached: bool = True, no_deps: bool = False
+) -> list[str]:
     # TODO: double check -e , --add-host, -v, --read-only
     dirname = compose.dirname
     name = cnt["name"]
@@ -1301,7 +1352,7 @@ class ServiceDependencyCondition(Enum):
     UNHEALTHY = "unhealthy"
 
     @classmethod
-    def from_value(cls, value):
+    def from_value(cls, value: str) -> 'ServiceDependencyCondition':
         # Check if the value exists in the enum
         for member in cls:
             if member.value == value:
@@ -1321,30 +1372,32 @@ class ServiceDependencyCondition(Enum):
 
 
 class ServiceDependency:
-    def __init__(self, name, condition):
+    def __init__(self, name: str, condition: str) -> None:
         self._name = name
         self._condition = ServiceDependencyCondition.from_value(condition)
 
     @property
-    def name(self):
+    def name(self) -> str:
         return self._name
 
     @property
-    def condition(self):
+    def condition(self) -> 'ServiceDependencyCondition':
         return self._condition
 
-    def __hash__(self):
+    def __hash__(self) -> int:
         # Compute hash based on the frozenset of items to ensure order does not matter
         return hash(('name', self._name) + ('condition', self._condition))
 
-    def __eq__(self, other):
+    def __eq__(self, other: Any) -> bool:
         # Compare equality based on dictionary content
         if isinstance(other, ServiceDependency):
             return self._name == other.name and self._condition == other.condition
         return False
 
 
-def rec_deps(services, service_name, start_point=None):
+def rec_deps(
+    services: dict[str, Any], service_name: str, start_point: str | None = None
+) -> set['ServiceDependency']:
     """
     return all dependencies of service_name recursively
     """
@@ -1366,13 +1419,13 @@ def rec_deps(services, service_name, start_point=None):
     return deps
 
 
-def flat_deps(services, with_extends=False):
+def flat_deps(services: dict[str, Any], with_extends: bool = False) -> None:
     """
     create dependencies "_deps" or update it recursively for all services
     """
     for name, srv in services.items():
         # parse dependencies for each service
-        deps = set()
+        deps: set[ServiceDependency] = set()
         srv["_deps"] = deps
         # TODO: manage properly the dependencies coming from base services when extended
         if with_extends:
@@ -1414,7 +1467,8 @@ class OverrideTag(yaml.YAMLObject):
     yaml_loader = yaml.SafeLoader
     yaml_tag = '!override'
 
-    def __init__(self, value):
+    def __init__(self, value: Any) -> None:
+        self.value: Union[dict[Any, Any], list[Any]]  # type: ignore[no-redef]
         if len(value) > 0 and isinstance(value[0], tuple):
             self.value = {}
             # item is a tuple representing service's lower level key and value
@@ -1422,18 +1476,18 @@ class OverrideTag(yaml.YAMLObject):
                 # value can actually be a list, then all the elements from the list have to be
                 # collected
                 if isinstance(item[1].value, list):
-                    self.value[item[0].value] = [item.value for item in item[1].value]
+                    self.value[item[0].value] = [i.value for i in item[1].value]  # type: ignore[index]
                 else:
-                    self.value[item[0].value] = item[1].value
+                    self.value[item[0].value] = item[1].value  # type: ignore[index]
         else:
-            self.value = [item.value for item in value]
+            self.value = [item.value for item in value]  # type: ignore[union-attr]
 
     @classmethod
-    def from_yaml(cls, loader, node):
+    def from_yaml(cls, loader: Any, node: Any) -> 'OverrideTag':
         return OverrideTag(node.value)
 
     @classmethod
-    def to_yaml(cls, dumper, data):
+    def to_yaml(cls, dumper: Any, data: 'OverrideTag') -> str:
         return dumper.represent_scalar(cls.yaml_tag, data.value)
 
 
@@ -1443,19 +1497,19 @@ class ResetTag(yaml.YAMLObject):
     yaml_tag = '!reset'
 
     @classmethod
-    def to_json(cls):
+    def to_json(cls) -> str:
         return cls.yaml_tag
 
     @classmethod
-    def from_yaml(cls, loader, node):
+    def from_yaml(cls, loader: Any, node: Any) -> 'ResetTag':
         return ResetTag()
 
     @classmethod
-    def to_yaml(cls, dumper, data):
+    def to_yaml(cls, dumper: Any, data: 'ResetTag') -> str:
         return dumper.represent_scalar(cls.yaml_tag, '')
 
 
-async def wait_with_timeout(coro, timeout):
+async def wait_with_timeout(coro: Any, timeout: int | float) -> Any:
     """
     Asynchronously waits for the given coroutine to complete with a timeout.
 
@@ -1480,17 +1534,19 @@ async def wait_with_timeout(coro, timeout):
 class Podman:
     def __init__(
         self,
-        compose,
-        podman_path="podman",
-        dry_run=False,
+        compose: PodmanCompose,
+        podman_path: str = "podman",
+        dry_run: bool = False,
         semaphore: asyncio.Semaphore = asyncio.Semaphore(sys.maxsize),
-    ):
+    ) -> None:
         self.compose = compose
         self.podman_path = podman_path
         self.dry_run = dry_run
         self.semaphore = semaphore
 
-    async def output(self, podman_args, cmd="", cmd_args=None):
+    async def output(
+        self, podman_args: list[str], cmd: str = "", cmd_args: list[str] | None = None
+    ) -> bytes:
         async with self.semaphore:
             cmd_args = cmd_args or []
             xargs = self.compose.get_podman_args(cmd) if cmd else []
@@ -1501,12 +1557,13 @@ class Podman:
             )
 
             stdout_data, stderr_data = await p.communicate()
+            assert p.returncode is not None
             if p.returncode == 0:
                 return stdout_data
 
             raise subprocess.CalledProcessError(p.returncode, " ".join(cmd_ls), stderr_data)
 
-    async def _readchunk(self, reader):
+    async def _readchunk(self, reader: asyncio.StreamReader) -> bytes:
         try:
             return await reader.readuntil(b"\n")
         except asyncio.exceptions.IncompleteReadError as e:
@@ -1514,16 +1571,18 @@ class Podman:
         except asyncio.exceptions.LimitOverrunError as e:
             return await reader.read(e.consumed)
 
-    async def _format_stream(self, reader, sink, log_formatter):
+    async def _format_stream(
+        self, reader: asyncio.StreamReader, sink: Any, log_formatter: str
+    ) -> None:
         line_ongoing = False
 
-        def _formatted_print_with_nl(s):
+        def _formatted_print_with_nl(s: str) -> None:
             if line_ongoing:
                 print(s, file=sink, end="\n")
             else:
                 print(log_formatter, s, file=sink, end="\n")
 
-        def _formatted_print_without_nl(s):
+        def _formatted_print_without_nl(s: str) -> None:
             if line_ongoing:
                 print(s, file=sink, end="")
             else:
@@ -1547,10 +1606,10 @@ class Podman:
 
     def exec(
         self,
-        podman_args,
-        cmd="",
-        cmd_args=None,
-    ):
+        podman_args: list[str],
+        cmd: str = "",
+        cmd_args: list[str] | None = None,
+    ) -> None:
         cmd_args = list(map(str, cmd_args or []))
         xargs = self.compose.get_podman_args(cmd) if cmd else []
         cmd_ls = [self.podman_path, *podman_args, cmd] + xargs + cmd_args
@@ -1559,14 +1618,14 @@ class Podman:
 
     async def run(  # pylint: disable=dangerous-default-value
         self,
-        podman_args,
-        cmd="",
-        cmd_args=None,
-        log_formatter=None,
+        podman_args: list[str],
+        cmd: str = "",
+        cmd_args: list[str] | None = None,
+        log_formatter: str | None = None,
         *,
         # Intentionally mutable default argument to hold references to tasks
-        task_reference=set(),
-    ) -> int:
+        task_reference: set[asyncio.Task] = set(),
+    ) -> int | None:
         async with self.semaphore:
             cmd_args = list(map(str, cmd_args or []))
             xargs = self.compose.get_podman_args(cmd) if cmd else []
@@ -1583,6 +1642,9 @@ class Podman:
                     close_fds=False,
                 )  # pylint: disable=consider-using-with
 
+                assert p.stdout is not None
+                assert p.stderr is not None
+
                 # This is hacky to make the tasks not get garbage collected
                 # https://github.com/python/cpython/issues/91887
                 out_t = asyncio.create_task(
@@ -1615,7 +1677,7 @@ class Podman:
             log.info("exit code: %s", exit_code)
             return exit_code
 
-    async def network_ls(self):
+    async def network_ls(self) -> list[str]:
         output = (
             await self.output(
                 [],
@@ -1633,7 +1695,7 @@ class Podman:
         networks = output.splitlines()
         return networks
 
-    async def volume_ls(self):
+    async def volume_ls(self) -> list[str]:
         output = (
             await self.output(
                 [],
@@ -1652,12 +1714,12 @@ class Podman:
         return volumes
 
 
-def normalize_service(service, sub_dir=""):
+def normalize_service(service: dict[str, Any], sub_dir: str = "") -> dict[str, Any]:
     if isinstance(service, ResetTag):
         return service
 
     if isinstance(service, OverrideTag):
-        service = service.value
+        service = service.value  # type: ignore[assignment]
 
     if "build" in service:
         build = service["build"]
@@ -1725,7 +1787,7 @@ def normalize_service(service, sub_dir=""):
     return service
 
 
-def normalize(compose):
+def normalize(compose: dict[str, Any]) -> dict[str, Any]:
     """
     convert compose dict of some keys from string or dicts into arrays
     """
@@ -1735,7 +1797,7 @@ def normalize(compose):
     return compose
 
 
-def normalize_service_final(service: dict, project_dir: str) -> dict:
+def normalize_service_final(service: dict[str, Any], project_dir: str) -> dict[str, Any]:
     if "build" in service:
         build = service["build"]
         context = build if isinstance(build, str) else build.get("context", ".")
@@ -1748,18 +1810,18 @@ def normalize_service_final(service: dict, project_dir: str) -> dict:
     return service
 
 
-def normalize_final(compose: dict, project_dir: str) -> dict:
+def normalize_final(compose: dict[str, Any], project_dir: str) -> dict[str, Any]:
     services = compose.get("services", {})
     for service in services.values():
         normalize_service_final(service, project_dir)
     return compose
 
 
-def clone(value):
+def clone(value: Any) -> Any:
     return value.copy() if is_list(value) or isinstance(value, dict) else value
 
 
-def rec_merge_one(target, source):
+def rec_merge_one(target: dict[str, Any], source: dict[str, Any]) -> dict[str, Any]:
     """
     update target from source recursively
     """
@@ -1829,7 +1891,7 @@ def rec_merge_one(target, source):
     return target
 
 
-def rec_merge(target, *sources):
+def rec_merge(target: dict[str, Any], *sources: dict[str, Any]) -> dict[str, Any]:
     """
     update target recursively from sources
     """
@@ -1838,7 +1900,9 @@ def rec_merge(target, *sources):
     return ret
 
 
-def resolve_extends(services, service_names, environ):
+def resolve_extends(
+    services: dict[str, Any], service_names: list[str], environ: dict[str, Any]
+) -> None:
     for name in service_names:
         service = services[name]
         ext = service.get("extends", {})
@@ -1870,7 +1934,7 @@ def resolve_extends(services, service_names, environ):
         services[name] = new_service
 
 
-def dotenv_to_dict(dotenv_path):
+def dotenv_to_dict(dotenv_path: str) -> dict[str, str | None]:
     if not os.path.isfile(dotenv_path):
         return {}
     return dotenv_values(dotenv_path)
@@ -1895,28 +1959,28 @@ COMPOSE_DEFAULT_LS = [
 
 
 class PodmanCompose:
-    def __init__(self):
+    def __init__(self) -> None:
         self.podman: Podman
-        self.podman_version = None
-        self.environ = {}
+        self.podman_version: str | None = None
+        self.environ: dict[str, str] = {}
         self.exit_code = None
-        self.commands = {}
+        self.commands: dict[str, Any] = {}
         self.global_args = argparse.Namespace()
-        self.project_name = None
-        self.dirname = None
-        self.pods = None
-        self.containers = []
-        self.vols = None
-        self.networks = {}
-        self.default_net = "default"
-        self.declared_secrets = None
-        self.container_names_by_service = None
-        self.container_by_name = None
-        self.services = None
-        self.all_services = set()
+        self.project_name: str | None = None
+        self.dirname: str
+        self.pods: list[Any]
+        self.containers: list[Any] = []
+        self.vols: dict[str, Any] | None = None
+        self.networks: dict[str, Any] = {}
+        self.default_net: str | None = "default"
+        self.declared_secrets: dict[str, Any] | None = None
+        self.container_names_by_service: dict[str, list[str]]
+        self.container_by_name: dict[str, Any]
+        self.services: dict[str, Any]
+        self.all_services: set[Any] = set()
         self.prefer_volume_over_mount = True
-        self.x_podman = {}
-        self.merged_yaml = None
+        self.x_podman: dict[str, Any] = {}
+        self.merged_yaml: Any
         self.yaml_hash = ""
         self.console_colors = [
             "\x1b[1;32m",
@@ -1926,7 +1990,7 @@ class PodmanCompose:
             "\x1b[1;36m",
         ]
 
-    def assert_services(self, services):
+    def assert_services(self, services: dict[str, Any]) -> None:
         if isinstance(services, str):
             services = [services]
         given = set(services or [])
@@ -1936,7 +2000,7 @@ class PodmanCompose:
             log.warning("missing services [%s]", missing_csv)
             sys.exit(1)
 
-    def get_podman_args(self, cmd):
+    def get_podman_args(self, cmd: str) -> list[str]:
         xargs = []
         for args in self.global_args.podman_args:
             xargs.extend(shlex.split(args))
@@ -1946,7 +2010,7 @@ class PodmanCompose:
             xargs.extend(shlex.split(args))
         return xargs
 
-    async def run(self, argv=None):
+    async def run(self, argv: list[str] | None = None) -> None:
         log.info("podman-compose version: %s", __version__)
         args = self._parse_args(argv)
         podman_path = args.podman_path
@@ -1984,13 +2048,13 @@ class PodmanCompose:
         if isinstance(retcode, int):
             sys.exit(retcode)
 
-    def resolve_in_pod(self):
+    def resolve_in_pod(self) -> bool:
         if self.global_args.in_pod in (None, ''):
             self.global_args.in_pod = self.x_podman.get("in_pod", "1")
         # otherwise use `in_pod` value provided by command line
         return self.global_args.in_pod
 
-    def resolve_pod_args(self):
+    def resolve_pod_args(self) -> list[str]:
         # Priorities:
         # - Command line --pod-args
         # - docker-compose.yml x-podman.pod_args
@@ -1999,7 +2063,7 @@ class PodmanCompose:
             return shlex.split(self.global_args.pod_args)
         return self.x_podman.get("pod_args", ["--infra=false", "--share="])
 
-    def _parse_compose_file(self):
+    def _parse_compose_file(self) -> None:
         args = self.global_args
         # cmd = args.command
         project_dir = os.environ.get("COMPOSE_PROJECT_DIR")
@@ -2033,7 +2097,7 @@ class PodmanCompose:
         # no_cleanup = args.no_cleanup
         # dry_run = args.dry_run
         # host_env = None
-        dirname = os.path.realpath(os.path.dirname(filename))
+        dirname: str = os.path.realpath(os.path.dirname(filename))
         dir_basename = os.path.basename(dirname)
         self.dirname = dirname
 
@@ -2050,9 +2114,11 @@ class PodmanCompose:
             dotenv_dict.update(dotenv_to_dict(dotenv_path))
 
         os.environ.update({
-            key: value for key, value in dotenv_dict.items() if key.startswith("PODMAN_")
+            key: value  # type: ignore[misc]
+            for key, value in dotenv_dict.items()
+            if key.startswith("PODMAN_")  # type: ignore[misc]
         })
-        self.environ = dotenv_dict
+        self.environ = dotenv_dict  # type: ignore[assignment]
         self.environ.update(dict(os.environ))
         # see: https://docs.docker.com/compose/reference/envvars/
         # see: https://docs.docker.com/compose/env-file/
@@ -2064,9 +2130,9 @@ class PodmanCompose:
 
         if args and 'env' in args and args.env:
             env_vars = norm_as_dict(args.env)
-            self.environ.update(env_vars)
+            self.environ.update(env_vars)  # type: ignore[arg-type]
 
-        compose = {}
+        compose: dict[str, Any] = {}
         # Iterate over files primitively to allow appending to files in-loop
         files_iter = iter(files)
 
@@ -2106,11 +2172,12 @@ class PodmanCompose:
                 project_name = project_name_normalized
 
             self.project_name = project_name
+            assert self.project_name is not None
             self.environ.update({"COMPOSE_PROJECT_NAME": self.project_name})
 
             content = rec_subs(content, self.environ)
-            if isinstance(services := content.get('services'), dict):
-                for service in services.values():
+            if isinstance(content_services := content.get('services'), dict):
+                for service in content_services.values():
                     if not isinstance(service, OverrideTag) and not isinstance(service, ResetTag):
                         if 'extends' in service and (
                             service_file := service['extends'].get('file')
@@ -2143,7 +2210,7 @@ class PodmanCompose:
             log.debug(" ** merged:\n%s", json.dumps(compose, indent=2))
         # ver = compose.get('version')
 
-        services = compose.get("services")
+        services: dict | None = compose.get("services")
         if services is None:
             services = {}
             log.warning("WARNING: No services defined")
@@ -2153,8 +2220,7 @@ class PodmanCompose:
         # NOTE: maybe add "extends.service" to _deps at this stage
         flat_deps(services, with_extends=True)
         service_names = sorted([(len(srv["_deps"]), name) for name, srv in services.items()])
-        service_names = [name for _, name in service_names]
-        resolve_extends(services, service_names, self.environ)
+        resolve_extends(services, [name for _, name in service_names], self.environ)
         flat_deps(services)
         nets = compose.get("networks", {})
         if not nets:
@@ -2208,7 +2274,7 @@ class PodmanCompose:
         # configs: {...}
         self.declared_secrets = compose.get("secrets", {})
         given_containers = []
-        container_names_by_service = {}
+        container_names_by_service: dict[str, list[str]] = {}
         self.services = services
         for service_name, service_desc in services.items():
             replicas = 1
@@ -2265,7 +2331,7 @@ class PodmanCompose:
                     if (
                         mnt_dict.get("type") == "volume"
                         and mnt_dict["source"]
-                        and mnt_dict["source"] not in self.vols
+                        and mnt_dict["source"] not in self.vols  # type: ignore[operator]
                     ):
                         vol_name = mnt_dict["source"]
                         raise RuntimeError(f"volume [{vol_name}] not defined in top level")
@@ -2286,7 +2352,9 @@ class PodmanCompose:
         self.containers = containers
         self.container_by_name = {c["name"]: c for c in containers}
 
-    def _resolve_profiles(self, defined_services, requested_profiles=None):
+    def _resolve_profiles(
+        self, defined_services: dict[str, Any], requested_profiles: set[str] | None = None
+    ) -> dict[str, Any]:
         """
         Returns a service dictionary (key = service name, value = service config) compatible with
         the requested_profiles list.
@@ -2308,7 +2376,7 @@ class PodmanCompose:
                 services[name] = config
         return services
 
-    def _parse_args(self, argv=None):
+    def _parse_args(self, argv: list[str] | None = None) -> argparse.Namespace:
         parser = argparse.ArgumentParser(formatter_class=argparse.RawTextHelpFormatter)
         self._init_global_parser(parser)
         subparsers = parser.add_subparsers(title="command", dest="command")
@@ -2329,7 +2397,7 @@ class PodmanCompose:
         return self.global_args
 
     @staticmethod
-    def _init_global_parser(parser):
+    def _init_global_parser(parser: argparse.ArgumentParser) -> None:
         parser.add_argument("-v", "--version", help="show version", action="store_true")
         parser.add_argument(
             "--in-pod",
@@ -2436,32 +2504,32 @@ class PodmanComposeError(Exception):
 
 
 class cmd_run:  # pylint: disable=invalid-name,too-few-public-methods
-    def __init__(self, compose, cmd_name, cmd_desc=None):
+    def __init__(self, compose: PodmanCompose, cmd_name: str, cmd_desc: str | None = None) -> None:
         self.compose = compose
         self.cmd_name = cmd_name
         self.cmd_desc = cmd_desc
 
-    def __call__(self, func):
-        def wrapped(*args, **kw):
+    def __call__(self, func: Callable) -> Callable:
+        def wrapped(*args: Any, **kw: Any) -> Any:
             return func(*args, **kw)
 
         if not asyncio.iscoroutinefunction(func):
             raise PodmanComposeError("Command must be async")
-        wrapped._compose = self.compose
+        wrapped._compose = self.compose  # type: ignore[attr-defined]
         # Trim extra indentation at start of multiline docstrings.
-        wrapped.desc = self.cmd_desc or re.sub(r"^\s+", "", func.__doc__)
-        wrapped._parse_args = []
+        wrapped.desc = self.cmd_desc or re.sub(r"^\s+", "", func.__doc__ or "")  # type: ignore[attr-defined]
+        wrapped._parse_args = []  # type: ignore[attr-defined]
         self.compose.commands[self.cmd_name] = wrapped
         return wrapped
 
 
 class cmd_parse:  # pylint: disable=invalid-name,too-few-public-methods
-    def __init__(self, compose, cmd_names):
+    def __init__(self, compose: PodmanCompose, cmd_names: list[str] | str) -> None:
         self.compose = compose
-        self.cmd_names = cmd_names if is_list(cmd_names) else [cmd_names]
+        self.cmd_names = cmd_names if is_list(cmd_names) else [cmd_names]  # type: ignore[list-item]
 
-    def __call__(self, func):
-        def wrapped(*args, **kw):
+    def __call__(self, func: Callable) -> Callable:
+        def wrapped(*args: Any, **kw: Any) -> Any:
             return func(*args, **kw)
 
         for cmd_name in self.cmd_names:
@@ -2475,7 +2543,7 @@ class cmd_parse:  # pylint: disable=invalid-name,too-few-public-methods
 
 
 @cmd_run(podman_compose, "version", "show version")
-async def compose_version(compose, args):
+async def compose_version(compose: PodmanCompose, args: argparse.Namespace) -> None:
     if getattr(args, "short", False):
         print(__version__)
         return
@@ -2500,7 +2568,7 @@ def is_local(container: dict) -> bool:
 
 
 @cmd_run(podman_compose, "wait", "wait running containers to stop")
-async def compose_wait(compose, args):  # pylint: disable=unused-argument
+async def compose_wait(compose: PodmanCompose, args: argparse.Namespace) -> None:
     containers = [cnt["name"] for cnt in compose.containers]
     cmd_args = ["--"]
     cmd_args.extend(containers)
@@ -2508,7 +2576,7 @@ async def compose_wait(compose, args):  # pylint: disable=unused-argument
 
 
 @cmd_run(podman_compose, "systemd")
-async def compose_systemd(compose, args):
+async def compose_systemd(compose: PodmanCompose, args: argparse.Namespace) -> None:
     """
     create systemd unit file and register its compose stacks
 
@@ -2613,7 +2681,7 @@ while in your project type `podman-compose systemd -a register`
 
 
 @cmd_run(podman_compose, "pull", "pull stack images")
-async def compose_pull(compose, args):
+async def compose_pull(compose: PodmanCompose, args: argparse.Namespace) -> None:
     img_containers = [cnt for cnt in compose.containers if "image" in cnt]
     if args.services:
         services = set(args.services)
@@ -2627,7 +2695,7 @@ async def compose_pull(compose, args):
 
 
 @cmd_run(podman_compose, "push", "push stack images")
-async def compose_push(compose, args):
+async def compose_push(compose: PodmanCompose, args: argparse.Namespace) -> None:
     services = set(args.services)
     for cnt in compose.containers:
         if "build" not in cnt:
@@ -2637,12 +2705,12 @@ async def compose_push(compose, args):
         await compose.podman.run([], "push", [cnt["image"]])
 
 
-def is_path_git_url(path):
+def is_path_git_url(path: str) -> bool:
     r = urllib.parse.urlparse(path)
     return r.scheme == 'git' or r.path.endswith('.git')
 
 
-def adjust_build_ssh_key_paths(compose, agent_or_key):
+def adjust_build_ssh_key_paths(compose: PodmanCompose, agent_or_key: str) -> str:
     # when using a custom id for ssh property, path to a local SSH key is provided after "="
     parts = agent_or_key.split("=", 1)
     if len(parts) == 1:
@@ -2652,7 +2720,13 @@ def adjust_build_ssh_key_paths(compose, agent_or_key):
     return name + "=" + os.path.join(compose.dirname, path)
 
 
-def container_to_build_args(compose, cnt, args, path_exists, cleanup_callbacks=None):
+def container_to_build_args(
+    compose: PodmanCompose,
+    cnt: dict[str, Any],
+    args: argparse.Namespace,
+    path_exists: Callable[[str], bool],
+    cleanup_callbacks: list[Callable] | None = None,
+) -> list[str]:
     build_desc = cnt["build"]
     if not hasattr(build_desc, "items"):
         build_desc = {"context": build_desc}
@@ -2669,7 +2743,7 @@ def container_to_build_args(compose, cnt, args, path_exists, cleanup_callbacks=N
         dockerfile.close()
         dockerfile = dockerfile.name
 
-        def cleanup_temp_dockfile():
+        def cleanup_temp_dockfile() -> None:
             if os.path.exists(dockerfile):
                 os.remove(dockerfile)
 
@@ -2743,7 +2817,7 @@ def container_to_build_args(compose, cnt, args, path_exists, cleanup_callbacks=N
     return build_args
 
 
-async def build_one(compose, args, cnt):
+async def build_one(compose: PodmanCompose, args: argparse.Namespace, cnt: dict) -> int | None:
     if "build" not in cnt:
         return None
     if getattr(args, "if_not_exists", None):
@@ -2756,7 +2830,7 @@ async def build_one(compose, args, cnt):
         if img_id:
             return None
 
-    cleanup_callbacks = []
+    cleanup_callbacks: list[Callable] = []
     build_args = container_to_build_args(
         compose, cnt, args, os.path.exists, cleanup_callbacks=cleanup_callbacks
     )
@@ -2767,7 +2841,7 @@ async def build_one(compose, args, cnt):
 
 
 @cmd_run(podman_compose, "build", "build stack images")
-async def compose_build(compose, args):
+async def compose_build(compose: PodmanCompose, args: argparse.Namespace) -> int:
     tasks = []
 
     if args.services:
@@ -2790,12 +2864,12 @@ async def compose_build(compose, args):
     return status
 
 
-async def pod_exists(compose, name):
+async def pod_exists(compose: PodmanCompose, name: str) -> bool:
     exit_code = await compose.podman.run([], "pod", ["exists", name])
     return exit_code == 0
 
 
-async def create_pods(compose, args):  # pylint: disable=unused-argument
+async def create_pods(compose: PodmanCompose, args: argparse.Namespace) -> None:
     for pod in compose.pods:
         if await pod_exists(compose, pod["name"]):
             continue
@@ -2814,7 +2888,7 @@ async def create_pods(compose, args):  # pylint: disable=unused-argument
         await compose.podman.run([], "pod", podman_args)
 
 
-def get_excluded(compose, args):
+def get_excluded(compose: PodmanCompose, args: argparse.Namespace) -> set[str]:
     excluded = set()
     if args.services:
         excluded = set(compose.services)
@@ -2839,7 +2913,10 @@ async def check_dep_conditions(compose: PodmanCompose, deps: set) -> None:
                 if (
                     d.condition
                     in (ServiceDependencyCondition.HEALTHY, ServiceDependencyCondition.UNHEALTHY)
-                ) and strverscmp_lt(compose.podman_version, "4.6.0"):
+                ) and (
+                    compose.podman_version is not None
+                    and strverscmp_lt(compose.podman_version, "4.6.0")
+                ):
                     log.warning(
                         "Ignored %s condition check due to podman %s doesn't support %s!",
                         d.name,
@@ -2877,8 +2954,8 @@ async def check_dep_conditions(compose: PodmanCompose, deps: set) -> None:
 
 
 async def run_container(
-    compose: PodmanCompose, name: str, deps: set, command: tuple, log_formatter: str = None
-):
+    compose: PodmanCompose, name: str, deps: set, command: tuple, log_formatter: str | None = None
+) -> int | None:
     """runs a container after waiting for its dependencies to be fulfilled"""
 
     # wait for the dependencies to be fulfilled
@@ -2888,17 +2965,17 @@ async def run_container(
 
     # start the container
     log.debug("Starting task for container %s", name)
-    return await compose.podman.run(*command, log_formatter=log_formatter)
+    return await compose.podman.run(*command, log_formatter=log_formatter or "text")  # type: ignore[misc]
 
 
-def deps_from_container(args, cnt):
+def deps_from_container(args: argparse.Namespace, cnt: dict) -> set:
     if args.no_deps:
         return set()
     return cnt['_deps']
 
 
 @cmd_run(podman_compose, "up", "Create and start the entire stack or some of its services")
-async def compose_up(compose: PodmanCompose, args):
+async def compose_up(compose: PodmanCompose, args: argparse.Namespace) -> int | None:
     excluded = get_excluded(compose, args)
 
     if not args.no_build:
@@ -2971,9 +3048,9 @@ async def compose_up(compose: PodmanCompose, args):
         curr_length = len(cnt["_service"])
         max_service_length = curr_length if curr_length > max_service_length else max_service_length
 
-    tasks = set()
+    tasks: set[asyncio.Task] = set()
 
-    async def handle_sigint():
+    async def handle_sigint() -> None:
         log.info("Caught SIGINT or Ctrl+C, shutting down...")
         try:
             log.info("Shutting down gracefully, please wait...")
@@ -3057,7 +3134,7 @@ async def compose_up(compose: PodmanCompose, args):
     return exit_code
 
 
-def get_volume_names(compose, cnt):
+def get_volume_names(compose: PodmanCompose, cnt: dict) -> list[str]:
     basedir = compose.dirname
     srv_name = cnt["_service"]
     ls = []
@@ -3074,9 +3151,9 @@ def get_volume_names(compose, cnt):
 
 
 @cmd_run(podman_compose, "down", "tear down entire stack")
-async def compose_down(compose: PodmanCompose, args):
+async def compose_down(compose: PodmanCompose, args: argparse.Namespace) -> None:
     excluded = get_excluded(compose, args)
-    podman_args = []
+    podman_args: list[str] = []
     timeout_global = getattr(args, "timeout", None)
     containers = list(reversed(compose.containers))
 
@@ -3159,7 +3236,7 @@ async def compose_down(compose: PodmanCompose, args):
 
 
 @cmd_run(podman_compose, "ps", "show status of containers")
-async def compose_ps(compose, args):
+async def compose_ps(compose: PodmanCompose, args: argparse.Namespace) -> None:
     ps_args = ["-a", "--filter", f"label=io.podman.compose.project={compose.project_name}"]
     if args.quiet is True:
         ps_args.extend(["--format", "{{.ID}}"])
@@ -3178,7 +3255,7 @@ async def compose_ps(compose, args):
     "run",
     "create a container similar to a service to run a one-off command",
 )
-async def compose_run(compose, args):
+async def compose_run(compose: PodmanCompose, args: argparse.Namespace) -> None:
     await create_pods(compose, args)
     compose.assert_services(args.service)
     container_names = compose.container_names_by_service[args.service]
@@ -3220,7 +3297,9 @@ async def compose_run(compose, args):
     sys.exit(p)
 
 
-def compose_run_update_container_from_args(compose, cnt, args):
+def compose_run_update_container_from_args(
+    compose: PodmanCompose, cnt: dict, args: argparse.Namespace
+) -> None:
     # adjust one-off container options
     name0 = "{}_{}_tmp{}".format(compose.project_name, args.service, random.randrange(0, 65536))
     cnt["name"] = args.name or name0
@@ -3259,7 +3338,7 @@ def compose_run_update_container_from_args(compose, cnt, args):
 
 
 @cmd_run(podman_compose, "exec", "execute a command in a running container")
-async def compose_exec(compose, args):
+async def compose_exec(compose: PodmanCompose, args: argparse.Namespace) -> None:
     compose.assert_services(args.service)
     container_names = compose.container_names_by_service[args.service]
     container_name = container_names[args.index - 1]
@@ -3269,7 +3348,7 @@ async def compose_exec(compose, args):
     sys.exit(p)
 
 
-def compose_exec_args(cnt, container_name, args):
+def compose_exec_args(cnt: dict, container_name: str, args: argparse.Namespace) -> list[str]:
     podman_args = ["--interactive"]
     if args.privileged:
         podman_args += ["--privileged"]
@@ -3293,7 +3372,9 @@ def compose_exec_args(cnt, container_name, args):
     return podman_args
 
 
-async def transfer_service_status(compose, args, action):
+async def transfer_service_status(
+    compose: PodmanCompose, args: argparse.Namespace, action: str
+) -> None:
     # TODO: handle dependencies, handle creations
     container_names_by_service = compose.container_names_by_service
     if not args.services:
@@ -3324,22 +3405,22 @@ async def transfer_service_status(compose, args, action):
 
 
 @cmd_run(podman_compose, "start", "start specific services")
-async def compose_start(compose, args):
+async def compose_start(compose: PodmanCompose, args: argparse.Namespace) -> None:
     await transfer_service_status(compose, args, "start")
 
 
 @cmd_run(podman_compose, "stop", "stop specific services")
-async def compose_stop(compose, args):
+async def compose_stop(compose: PodmanCompose, args: argparse.Namespace) -> None:
     await transfer_service_status(compose, args, "stop")
 
 
 @cmd_run(podman_compose, "restart", "restart specific services")
-async def compose_restart(compose, args):
+async def compose_restart(compose: PodmanCompose, args: argparse.Namespace) -> None:
     await transfer_service_status(compose, args, "restart")
 
 
 @cmd_run(podman_compose, "logs", "show logs from services")
-async def compose_logs(compose, args):
+async def compose_logs(compose: PodmanCompose, args: argparse.Namespace) -> None:
     container_names_by_service = compose.container_names_by_service
     if not args.services and not args.latest:
         args.services = container_names_by_service.keys()
@@ -3370,7 +3451,7 @@ async def compose_logs(compose, args):
 
 
 @cmd_run(podman_compose, "config", "displays the compose file")
-async def compose_config(compose, args):
+async def compose_config(compose: PodmanCompose, args: argparse.Namespace) -> None:
     if args.services:
         for service in compose.services:
             if not args.quiet:
@@ -3381,7 +3462,7 @@ async def compose_config(compose, args):
 
 
 @cmd_run(podman_compose, "port", "Prints the public port for a port binding.")
-async def compose_port(compose, args):
+async def compose_port(compose: PodmanCompose, args: argparse.Namespace) -> None:
     compose.assert_services(args.service)
     containers = compose.container_names_by_service[args.service]
     output = await compose.podman.output([], "inspect", [containers[args.index - 1]])
@@ -3392,7 +3473,7 @@ async def compose_port(compose, args):
 
 
 @cmd_run(podman_compose, "pause", "Pause all running containers")
-async def compose_pause(compose, args):
+async def compose_pause(compose: PodmanCompose, args: argparse.Namespace) -> None:
     container_names_by_service = compose.container_names_by_service
     if not args.services:
         args.services = container_names_by_service.keys()
@@ -3403,7 +3484,7 @@ async def compose_pause(compose, args):
 
 
 @cmd_run(podman_compose, "unpause", "Unpause all running containers")
-async def compose_unpause(compose, args):
+async def compose_unpause(compose: PodmanCompose, args: argparse.Namespace) -> None:
     container_names_by_service = compose.container_names_by_service
     if not args.services:
         args.services = container_names_by_service.keys()
@@ -3414,7 +3495,7 @@ async def compose_unpause(compose, args):
 
 
 @cmd_run(podman_compose, "kill", "Kill one or more running containers with a specific signal")
-async def compose_kill(compose, args):
+async def compose_kill(compose: PodmanCompose, args: argparse.Namespace) -> None:
     # to ensure that the user did not execute the command by mistake
     if not args.services and not args.all:
         log.fatal(
@@ -3450,7 +3531,7 @@ async def compose_kill(compose, args):
     "stats",
     "Display percentage of CPU, memory, network I/O, block I/O and PIDs for services.",
 )
-async def compose_stats(compose, args):
+async def compose_stats(compose: PodmanCompose, args: argparse.Namespace) -> None:
     container_names_by_service = compose.container_names_by_service
     if not args.services:
         args.services = container_names_by_service.keys()
@@ -3477,7 +3558,7 @@ async def compose_stats(compose, args):
 
 
 @cmd_run(podman_compose, "images", "List images used by the created containers")
-async def compose_images(compose, args):
+async def compose_images(compose: PodmanCompose, args: argparse.Namespace) -> None:
     img_containers = [cnt for cnt in compose.containers if "image" in cnt]
     data = []
     if args.quiet is True:
@@ -3518,7 +3599,7 @@ async def compose_images(compose, args):
 
 
 @cmd_parse(podman_compose, "version")
-def compose_version_parse(parser):
+def compose_version_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-f",
         "--format",
@@ -3534,7 +3615,7 @@ def compose_version_parse(parser):
 
 
 @cmd_parse(podman_compose, "up")
-def compose_up_parse(parser):
+def compose_up_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-d",
         "--detach",
@@ -3628,7 +3709,7 @@ def compose_up_parse(parser):
 
 
 @cmd_parse(podman_compose, "down")
-def compose_down_parse(parser):
+def compose_down_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-v",
         "--volumes",
@@ -3654,7 +3735,7 @@ def compose_down_parse(parser):
 
 
 @cmd_parse(podman_compose, "run")
-def compose_run_parse(parser):
+def compose_run_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--build", action="store_true", help="Build images before starting containers."
     )
@@ -3733,7 +3814,7 @@ def compose_run_parse(parser):
 
 
 @cmd_parse(podman_compose, "exec")
-def compose_exec_parse(parser):
+def compose_exec_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-d",
         "--detach",
@@ -3784,7 +3865,7 @@ def compose_exec_parse(parser):
 
 
 @cmd_parse(podman_compose, ["down", "stop", "restart"])
-def compose_parse_timeout(parser):
+def compose_parse_timeout(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-t",
         "--timeout",
@@ -3795,7 +3876,7 @@ def compose_parse_timeout(parser):
 
 
 @cmd_parse(podman_compose, ["logs"])
-def compose_logs_parse(parser):
+def compose_logs_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-f",
         "--follow",
@@ -3829,7 +3910,7 @@ def compose_logs_parse(parser):
 
 
 @cmd_parse(podman_compose, "systemd")
-def compose_systemd_parse(parser):
+def compose_systemd_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-a",
         "--action",
@@ -3840,7 +3921,7 @@ def compose_systemd_parse(parser):
 
 
 @cmd_parse(podman_compose, "pull")
-def compose_pull_parse(parser):
+def compose_pull_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--force-local",
         action="store_true",
@@ -3851,7 +3932,7 @@ def compose_pull_parse(parser):
 
 
 @cmd_parse(podman_compose, "push")
-def compose_push_parse(parser):
+def compose_push_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--ignore-push-failures",
         action="store_true",
@@ -3861,12 +3942,12 @@ def compose_push_parse(parser):
 
 
 @cmd_parse(podman_compose, "ps")
-def compose_ps_parse(parser):
+def compose_ps_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("-q", "--quiet", help="Only display container IDs", action="store_true")
 
 
 @cmd_parse(podman_compose, ["build", "up"])
-def compose_build_up_parse(parser):
+def compose_build_up_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--pull",
         help="attempt to pull a newer version of the image",
@@ -3893,7 +3974,7 @@ def compose_build_up_parse(parser):
 
 
 @cmd_parse(podman_compose, ["build", "up", "down", "start", "stop", "restart"])
-def compose_build_parse(parser):
+def compose_build_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "services",
         metavar="services",
@@ -3904,7 +3985,7 @@ def compose_build_parse(parser):
 
 
 @cmd_parse(podman_compose, "config")
-def compose_config_parse(parser):
+def compose_config_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--no-normalize", help="Don't normalize compose model.", action="store_true"
     )
@@ -3920,7 +4001,7 @@ def compose_config_parse(parser):
 
 
 @cmd_parse(podman_compose, "port")
-def compose_port_parse(parser):
+def compose_port_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "--index",
         type=int,
@@ -3944,14 +4025,14 @@ def compose_port_parse(parser):
 
 
 @cmd_parse(podman_compose, ["pause", "unpause"])
-def compose_pause_unpause_parse(parser):
+def compose_pause_unpause_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "services", metavar="services", nargs="*", default=None, help="service names"
     )
 
 
 @cmd_parse(podman_compose, ["kill"])
-def compose_kill_parse(parser):
+def compose_kill_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "services", metavar="services", nargs="*", default=None, help="service names"
     )
@@ -3970,12 +4051,12 @@ def compose_kill_parse(parser):
 
 
 @cmd_parse(podman_compose, "images")
-def compose_images_parse(parser):
+def compose_images_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("-q", "--quiet", help="Only display images IDs", action="store_true")
 
 
 @cmd_parse(podman_compose, ["stats"])
-def compose_stats_parse(parser):
+def compose_stats_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "services", metavar="services", nargs="*", default=None, help="service names"
     )
@@ -3998,7 +4079,7 @@ def compose_stats_parse(parser):
 
 
 @cmd_parse(podman_compose, ["ps", "stats"])
-def compose_format_parse(parser):
+def compose_format_parse(parser: argparse.ArgumentParser) -> None:
     parser.add_argument(
         "-f",
         "--format",
@@ -4007,11 +4088,11 @@ def compose_format_parse(parser):
     )
 
 
-async def async_main():
+async def async_main() -> None:
     await podman_compose.run()
 
 
-def main():
+def main() -> None:
     asyncio.run(async_main())