diff --git a/Makefile b/Makefile index eb56753a..0478f123 100644 --- a/Makefile +++ b/Makefile @@ -76,7 +76,7 @@ docs: # -D language=en : define language as en # . : source directory # _build/html : target - cd _build/doctrees && python3 -m sphinx -T -E -W --keep-going -b html -d _build/doctrees -D language=en . _build/html + python3 -m sphinx -T -E -W --keep-going -b html -d _build/doctrees -D language=en _build/doctrees _build/html .PHONY: rpm rpm: ## Build rpm packages diff --git a/docs/source/conf.py b/docs/source/conf.py index c593d8b3..42ea49ef 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -39,7 +39,7 @@ # sphinx.ext.napoleon: Support for NumPy and Google style docstrings # sphinx.ext.viewcode: Add links to highlighted source code # isort: unique-list -extensions = ['sphinx.ext.autodoc', 'sphinx.ext.napoleon', 'sphinx.ext.viewcode'] +extensions = ['sphinx.ext.napoleon', 'sphinx.ext.autodoc', 'sphinx.ext.viewcode'] # Add any paths that contain templates here, relative to this directory. templates_path = ['_templates'] @@ -51,6 +51,9 @@ exclude_patterns = [ 'podman.api.*rst', 'podman.rst', + 'podman.version.rst', + 'podman.tlsconfig.rst', + 'podman.errors.rst', ] diff --git a/docs/source/index.rst b/docs/source/index.rst index 3fdbc6a8..7e0fa609 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -53,14 +53,18 @@ Example :glob: :hidden: + podman.domain.config* podman.domain.containers podman.domain.containers_manager podman.domain.images podman.domain.images_manager + podman.domain.ipam* podman.domain.events* + podman.domain.manager podman.domain.manifests* podman.domain.networks* podman.domain.pods* + podman.domain.registry_data* podman.domain.secrets* podman.domain.system* podman.domain.volume* diff --git a/podman/domain/containers.py b/podman/domain/containers.py index f9446fa1..1576a15b 100644 --- a/podman/domain/containers.py +++ b/podman/domain/containers.py @@ -82,7 +82,10 @@ def attach_socket(self, **kwargs): """ raise NotImplementedError() - def commit(self, repository: str = None, tag: str = None, **kwargs) -> Image: + def commit(self, + repository: str = None, + tag: str = None, + **kwargs) -> Image: """Save container to given repository. Args: @@ -140,7 +143,8 @@ def exec_run( environment: Union[Mapping[str, str], List[str]] = None, workdir: str = None, demux: bool = False, - ) -> Tuple[Optional[int], Union[Iterator[bytes], Any, Tuple[bytes, bytes]]]: + ) -> Tuple[Optional[int], Union[Iterator[bytes], Any, Tuple[bytes, + bytes]]]: """Run given command inside container and return results. Args: @@ -187,20 +191,24 @@ def exec_run( "WorkingDir": workdir, } # create the exec instance - response = self.client.post(f"/containers/{self.name}/exec", data=json.dumps(data)) + response = self.client.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( - f"/exec/{exec_id}/start", data=json.dumps({"Detach": detach, "Tty": tty}) - ) + start_resp = self.client.post(f"/exec/{exec_id}/start", + data=json.dumps({ + "Detach": detach, + "Tty": tty + })) start_resp.raise_for_status() # get and return exec information response = self.client.get(f"/exec/{exec_id}/json") response.raise_for_status() return response.json().get('ExitCode'), start_resp.content - def export(self, chunk_size: int = api.DEFAULT_CHUNK_SIZE) -> Iterator[bytes]: + def export(self, + chunk_size: int = api.DEFAULT_CHUNK_SIZE) -> Iterator[bytes]: """Download container's filesystem contents as a tar archive. Args: @@ -213,14 +221,17 @@ 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.client.get(f"/containers/{self.id}/export", + stream=True) response.raise_for_status() for out in response.iter_content(chunk_size=chunk_size): yield out def get_archive( - self, path: str, chunk_size: int = api.DEFAULT_CHUNK_SIZE + self, + path: str, + chunk_size: int = api.DEFAULT_CHUNK_SIZE ) -> Tuple[Iterable, Dict[str, Any]]: """Download a file or folder from the container's filesystem. @@ -232,7 +243,8 @@ 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.client.get(f"/containers/{self.id}/archive", + params={"path": [path]}) response.raise_for_status() stat = response.headers.get("x-docker-container-path-stat", None) @@ -255,7 +267,8 @@ 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.client.post(f"/containers/{self.id}/kill", + params={"signal": signal}) response.raise_for_status() def logs(self, **kwargs) -> Union[bytes, Iterator[bytes]]: @@ -286,7 +299,9 @@ 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.client.get(f"/containers/{self.id}/logs", + stream=stream, + params=params) response.raise_for_status() if stream: @@ -318,9 +333,9 @@ 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.client.put(f"/containers/{self.id}/archive", + params={"path": path}, + data=data) return response.ok def remove(self, **kwargs) -> None: @@ -344,7 +359,8 @@ 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.client.post(f"/containers/{self.id}/rename", + params={"name": name}) response.raise_for_status() self.attrs["Name"] = name # shortcut to avoid needing reload() @@ -360,7 +376,8 @@ 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.client.post(f"/containers/{self.id}/resize", + params=params) response.raise_for_status() def restart(self, **kwargs) -> None: @@ -374,7 +391,9 @@ 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.client.post(f"/containers/{self.id}/restart", + params=params, + **post_kwargs) response.raise_for_status() def start(self, **kwargs) -> None: @@ -384,13 +403,14 @@ def start(self, **kwargs) -> None: detach_keys: Override the key sequence for detaching a container (Podman only) """ response = self.client.post( - f"/containers/{self.id}/start", params={"detachKeys": kwargs.get("detach_keys")} - ) + f"/containers/{self.id}/start", + params={"detachKeys": kwargs.get("detach_keys")}) response.raise_for_status() def stats( self, **kwargs - ) -> Union[bytes, Dict[str, Any], Iterator[bytes], Iterator[Dict[str, Any]]]: + ) -> Union[bytes, Dict[str, Any], Iterator[bytes], Iterator[Dict[str, + Any]]]: """Return statistics for container. Keyword Args: @@ -410,7 +430,9 @@ def stats( "stream": stream, } - response = self.client.get("/containers/stats", params=params, stream=stream) + response = self.client.get("/containers/stats", + params=params, + stream=stream) response.raise_for_status() if stream: @@ -432,7 +454,9 @@ 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.client.post(f"/containers/{self.id}/stop", + params=params, + **post_kwargs) response.raise_for_status() if response.status_code == requests.codes.no_content: @@ -443,7 +467,9 @@ def stop(self, **kwargs) -> None: return body = response.json() - raise APIError(body["cause"], response=response, explanation=body["message"]) + raise APIError(body["cause"], + response=response, + explanation=body["message"]) def top(self, **kwargs) -> Union[Iterator[Dict[str, Any]], Dict[str, Any]]: """Report on running processes in the container. @@ -462,7 +488,9 @@ 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.client.get(f"/containers/{self.id}/top", + params=params, + stream=stream) response.raise_for_status() if stream: @@ -481,7 +509,8 @@ def update(self, **kwargs): Raises: NotImplementedError: Podman service unsupported operation. """ - raise NotImplementedError("Container.update() is not supported by Podman service.") + raise NotImplementedError( + "Container.update() is not supported by Podman service.") def wait(self, **kwargs) -> int: """Block until the container enters given state. @@ -515,6 +544,7 @@ def wait(self, **kwargs) -> int: # This API endpoint responds with a JSON encoded integer. # 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) + response = self.client.post(f"/containers/{self.id}/wait", + params=params) response.raise_for_status() return response.json() diff --git a/podman/domain/containers_create.py b/podman/domain/containers_create.py index d9724b0b..3bc50e95 100644 --- a/podman/domain/containers_create.py +++ b/podman/domain/containers_create.py @@ -18,9 +18,10 @@ class CreateMixin: # pylint: disable=too-few-public-methods """Class providing create method for ContainersManager.""" - def create( - self, image: Union[Image, str], command: Union[str, List[str], None] = None, **kwargs - ) -> Container: + def create(self, + image: Union[Image, str], + command: Union[str, List[str], None] = None, + **kwargs) -> Container: """Create a container. Args: @@ -104,33 +105,54 @@ def create( powerful alternative to volumes. Each item in the list is expected to be a Mount object. For example: - [ + + [ + { + "type": "bind", + "source": "/a/b/c1", + "target" "/d1", + "read_only": True, + "relabel": "Z" + }, + { + "type": "tmpfs", - "source": "tmpfs", # If this was not passed, the regular directory - # would be created rather than tmpfs mount !!! - # as this will cause to have invalid entry - # in /proc/self/mountinfo + + # If this was not passed, the regular directory + + # would be created rather than tmpfs mount !!! + + # as this will cause to have invalid entry + + # in /proc/self/mountinfo + + "source": "tmpfs", + "target" "/d2", + "size": "100k", + "chown": True + } - ] + ] name (str): The name for this container. nano_cpus (int): CPU quota in units of 1e-9 CPUs. networks (Dict[str, Dict[str, Union[str, List[str]]): Networks which will be connected to container during container creation Values of the network configuration can be : - - string - - list of strings (e.g. Aliases) + + - string + - list of strings (e.g. Aliases) network_disabled (bool): Disable networking. network_mode (str): One of: @@ -150,9 +172,7 @@ def create( pids_limit (int): Tune a container's pids limit. Set -1 for unlimited. platform (str): Platform in the format os[/arch[/variant]]. Only used if the method needs to pull the requested image. - ports (Dict[str, Union[int, Tuple[str, int], List[int], - Dict[str, Union[int, Tuple[str, int], List[int]]]]] - ): Ports to bind inside the container. + ports (Dict[str, Union[int, Tuple[str, int], List[int], Dict[str, Union[int, Tuple[str, int], List[int]]]]]): Ports to bind inside the container. The keys of the dictionary are the ports to bind inside the container, either as an integer or a string in the form port/protocol, where the protocol is either @@ -162,31 +182,40 @@ def create( which can be either: - The port number, as an integer. + For example: {'2222/tcp': 3333} will expose port 2222 inside the container as port 3333 on the host. - None, to assign a random host port. + For example: {'2222/tcp': None}. - A tuple of (address, port) if you want to specify the host interface. + For example: {'1111/tcp': ('127.0.0.1', 1111)}. - A list of integers or tuples of (address, port), if you want to bind multiple host ports to a single container port. + For example: {'1111/tcp': [1234, ("127.0.0.1", 4567)]}. For example: {'9090': 7878, '10932/tcp': '8781', "8989/tcp": ("127.0.0.1", 9091)} - A dictionary of the options mentioned above except for random host port. + The dictionary has additional option "range", which allows binding range of ports. For example: - - {'2222/tcp': {"port": 3333, "range": 4}} - - {'1111/tcp': {"port": ('127.0.0.1', 1111), "range": 4}} - - {'1111/tcp': [ - {"port": 1234, "range": 4}, - {"ip": "127.0.0.1", "port": 4567} - ] - } + - {'2222/tcp': {"port": 3333, "range": 4}} + - {'1111/tcp': {"port": ('127.0.0.1', 1111), "range": 4}} + - {'1111/tcp': [ + + {"port": 1234, "range": 4}, + + {"ip": "127.0.0.1", "port": 4567} + + ] + + } privileged (bool): Give extended privileges to this container. publish_all_ports (bool): Publish all ports to the host. read_only (bool): Mount the container's root filesystem as read only. @@ -200,34 +229,41 @@ def create( - MaximumRetryCount: Number of times to restart the container on failure. For example: {"Name": "on-failure", "MaximumRetryCount": 5} - runtime (str): Runtime to use with this container. secrets (List[Union[str, Secret, Dict[str, Union[str, int]]]]): Secrets to mount to this container. For example: - - As list of strings, each string representing a secret's ID or name: - ['my_secret', 'my_secret2'] - - - As list of Secret objects the corresponding IDs are read from: - [Secret, Secret] - - - As list of dictionaries: - [ - { - "source": "my_secret", # A string representing the ID or name of - # a secret - "target": "/my_secret", # An optional target to mount source to, - # falls back to /run/secrets/source - "uid": 1000, # An optional UID that falls back to 0 - # if not given - "gid": 1000, # An optional GID that falls back to 0 - # if not given - "mode": 0o400, # An optional mode to apply to the target, - # use an 0o prefix for octal integers - }, - ] + - As list of strings, each string representing a secret's ID or name: + ['my_secret', 'my_secret2'] + + - As list of Secret objects the corresponding IDs are read from: + [Secret, Secret] + + - As list of dictionaries: + [ + + { + + "source": "my_secret", # A string representing the ID or name of + # a secret + + "target": "/my_secret", # An optional target to mount source to, + # falls back to /run/secrets/source + + "uid": 1000, # An optional UID that falls back to 0 + # if not given + + "gid": 1000, # An optional GID that falls back to 0 + # if not given + + "mode": 0o400, # An optional mode to apply to the target, + # use an 0o prefix for octal integers + + }, + + ] secret_env (Dict[str, str]): Secrets to add as environment variables available in the container. @@ -257,11 +293,9 @@ def create( the corresponding environment variables will be set in the container being built. user (Union[str, int]): Username or UID to run commands as inside the container. userns_mode (str): Sets the user namespace mode for the container when user namespace - remapping option is enabled. Supported values documented here - https://docs.podman.io/en/latest/markdown/options/userns.container.html#userns-mode + remapping option is enabled. Supported values documented `here `_ uts_mode (str): Sets the UTS namespace mode for the container. - Supported values are documented here - https://docs.podman.io/en/latest/markdown/options/uts.container.html + `These `_ are the supported values. version (str): The version of the API to use. Set to auto to automatically detect the server's version. Default: 3.0.0 volume_driver (str): The name of a volume driver/plugin. @@ -278,12 +312,19 @@ def create( For example: { + 'test_bind_1': + {'bind': '/mnt/vol1', 'mode': 'rw'}, + 'test_bind_2': + {'bind': '/mnt/vol2', 'extended_mode': ['ro', 'noexec']}, + 'test_bind_3': + {'bind': '/mnt/vol3', 'extended_mode': ['noexec'], 'mode': 'rw'} + } volumes_from (List[str]): List of container names or IDs to get volumes from. @@ -305,8 +346,9 @@ def create( payload = api.prepare_body(payload) response = self.client.post( - "/containers/create", headers={"content-type": "application/json"}, data=payload - ) + "/containers/create", + headers={"content-type": "application/json"}, + data=payload) response.raise_for_status(not_found=ImageNotFound) container_id = response.json()["Id"] @@ -321,47 +363,45 @@ def _render_payload(kwargs: MutableMapping[str, Any]) -> Dict[str, Any]: if "links" in args: if len(args["links"]) > 0: - raise ValueError("'links' are not supported by Podman service.") + raise ValueError( + "'links' are not supported by Podman service.") del args["links"] # Ignore these keywords for key in ( - "cpu_count", - "cpu_percent", - "nano_cpus", - "platform", # used by caller - "remove", # used by caller - "stderr", # used by caller - "stdout", # used by caller - "stream", # used by caller - "detach", # used by caller - "volume_driver", + "cpu_count", + "cpu_percent", + "nano_cpus", + "platform", # used by caller + "remove", # used by caller + "stderr", # used by caller + "stdout", # used by caller + "stream", # used by caller + "detach", # used by caller + "volume_driver", ): with suppress(KeyError): del args[key] # These keywords are not supported for various reasons. - unsupported_keys = set(args.keys()).intersection( - ( - "blkio_weight", - "blkio_weight_device", # FIXME In addition to device Major/Minor include path - "device_cgroup_rules", # FIXME Where to map for Podman API? - "device_read_bps", # FIXME In addition to device Major/Minor include path - "device_read_iops", # FIXME In addition to device Major/Minor include path - "device_requests", # FIXME In addition to device Major/Minor include path - "device_write_bps", # FIXME In addition to device Major/Minor include path - "device_write_iops", # FIXME In addition to device Major/Minor include path - "domainname", - "network_disabled", # FIXME Where to map for Podman API? - "storage_opt", # FIXME Where to map for Podman API? - "tmpfs", # FIXME Where to map for Podman API? - ) - ) + unsupported_keys = set(args.keys()).intersection(( + "blkio_weight", + "blkio_weight_device", # FIXME In addition to device Major/Minor include path + "device_cgroup_rules", # FIXME Where to map for Podman API? + "device_read_bps", # FIXME In addition to device Major/Minor include path + "device_read_iops", # FIXME In addition to device Major/Minor include path + "device_requests", # FIXME In addition to device Major/Minor include path + "device_write_bps", # FIXME In addition to device Major/Minor include path + "device_write_iops", # FIXME In addition to device Major/Minor include path + "domainname", + "network_disabled", # FIXME Where to map for Podman API? + "storage_opt", # FIXME Where to map for Podman API? + "tmpfs", # FIXME Where to map for Podman API? + )) if len(unsupported_keys) > 0: raise TypeError( f"""Keyword(s) '{" ,".join(unsupported_keys)}' are""" - f""" currently not supported by Podman API.""" - ) + f""" currently not supported by Podman API.""") def pop(k): return args.pop(k, None) @@ -391,30 +431,34 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: except ValueError as bad_size: mapping = {'b': 0, 'k': 1, 'm': 2, 'g': 3} mapping_regex = ''.join(mapping.keys()) - search = re.search(rf'^(\d+)([{mapping_regex}])$', size.lower()) + search = re.search(rf'^(\d+)([{mapping_regex}])$', + size.lower()) if search: - return int(search.group(1)) * (1024 ** mapping[search.group(2)]) + return int( + search.group(1)) * (1024**mapping[search.group(2)]) raise TypeError( f"Passed string size {size} should be in format\\d+[bBkKmMgG] (e.g. '100m')" ) from bad_size else: raise TypeError( f"Passed size {size} should be a type of unicode, str " - f"or int (found : {size_type})" - ) + f"or int (found : {size_type})") # Transform keywords into parameters params = { "annotations": pop("annotations"), # TODO document, podman only - "apparmor_profile": pop("apparmor_profile"), # TODO document, podman only + "apparmor_profile": + pop("apparmor_profile"), # TODO document, podman only "cap_add": pop("cap_add"), "cap_drop": pop("cap_drop"), "cgroup_parent": pop("cgroup_parent"), "cgroups_mode": pop("cgroups_mode"), # TODO document, podman only "cni_networks": [pop("network")], "command": args.pop("command", args.pop("cmd", None)), - "conmon_pid_file": pop("conmon_pid_file"), # TODO document, podman only - "containerCreateCommand": pop("containerCreateCommand"), # TODO document, podman only + "conmon_pid_file": + pop("conmon_pid_file"), # TODO document, podman only + "containerCreateCommand": + pop("containerCreateCommand"), # TODO document, podman only "devices": [], "dns_options": pop("dns_opt"), "dns_search": pop("dns_search"), @@ -425,14 +469,17 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: "expose": {}, "groups": pop("group_add"), "healthconfig": pop("healthcheck"), - "health_check_on_failure_action": pop("health_check_on_failure_action"), + "health_check_on_failure_action": + pop("health_check_on_failure_action"), "hostadd": [], "hostname": pop("hostname"), "httpproxy": pop("use_config_proxy"), "idmappings": pop("idmappings"), # TODO document, podman only "image": pop("image"), - "image_volume_mode": pop("image_volume_mode"), # TODO document, podman only - "image_volumes": pop("image_volumes"), # TODO document, podman only + "image_volume_mode": + pop("image_volume_mode"), # TODO document, podman only + "image_volumes": + pop("image_volumes"), # TODO document, podman only "init": pop("init"), "init_path": pop("init_path"), "isolation": pop("isolation"), @@ -443,18 +490,22 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: "mounts": [], "name": pop("name"), "namespace": pop("namespace"), # TODO What is this for? - "network_options": pop("network_options"), # TODO document, podman only + "network_options": + pop("network_options"), # TODO document, podman only "networks": pop("networks"), - "no_new_privileges": pop("no_new_privileges"), # TODO document, podman only + "no_new_privileges": + pop("no_new_privileges"), # TODO document, podman only "oci_runtime": pop("runtime"), "oom_score_adj": pop("oom_score_adj"), - "overlay_volumes": pop("overlay_volumes"), # TODO document, podman only + "overlay_volumes": + pop("overlay_volumes"), # TODO document, podman only "portmappings": [], "privileged": pop("privileged"), "procfs_opts": pop("procfs_opts"), # TODO document, podman only "publish_image_ports": pop("publish_all_ports"), "r_limits": [], - "raw_image_name": pop("raw_image_name"), # TODO document, podman only + "raw_image_name": + pop("raw_image_name"), # TODO document, podman only "read_only_filesystem": pop("read_only"), "read_write_tmpfs": pop("read_write_tmpfs"), "remove": args.pop("remove", args.pop("auto_remove", None)), @@ -462,8 +513,10 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: "rootfs": pop("rootfs"), "rootfs_propagation": pop("rootfs_propagation"), "sdnotifyMode": pop("sdnotifyMode"), # TODO document, podman only - "seccomp_policy": pop("seccomp_policy"), # TODO document, podman only - "seccomp_profile_path": pop("seccomp_profile_path"), # TODO document, podman only + "seccomp_policy": + pop("seccomp_policy"), # TODO document, podman only + "seccomp_profile_path": + pop("seccomp_profile_path"), # TODO document, podman only "secrets": [], # TODO document, podman only "selinux_opts": pop("security_opt"), "shm_size": to_bytes(pop("shm_size")), @@ -478,8 +531,10 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: "umask": pop("umask"), # TODO document, podman only "unified": pop("unified"), # TODO document, podman only "unmask": pop("unmasked_paths"), # TODO document, podman only - "use_image_hosts": pop("use_image_hosts"), # TODO document, podman only - "use_image_resolve_conf": pop("use_image_resolve_conf"), # TODO document, podman only + "use_image_hosts": + pop("use_image_hosts"), # TODO document, podman only + "use_image_resolve_conf": + pop("use_image_resolve_conf"), # TODO document, podman only "user": pop("user"), "version": pop("version"), "volumes": [], @@ -498,12 +553,16 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: params["hostadd"].append(f"{hostname}:{ip}") if "log_config" in args: - params["log_configuration"]["driver"] = args["log_config"].get("Type") + params["log_configuration"]["driver"] = args["log_config"].get( + "Type") if "Config" in args["log_config"]: - params["log_configuration"]["path"] = args["log_config"]["Config"].get("path") - params["log_configuration"]["size"] = args["log_config"]["Config"].get("size") - params["log_configuration"]["options"] = args["log_config"]["Config"].get("options") + params["log_configuration"]["path"] = args["log_config"][ + "Config"].get("path") + params["log_configuration"]["size"] = args["log_config"][ + "Config"].get("size") + params["log_configuration"]["options"] = args["log_config"][ + "Config"].get("options") args.pop("log_config") for item in args.pop("mounts", []): @@ -544,10 +603,14 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: def parse_host_port(_container_port, _protocol, _host): result = [] - port_map = {"container_port": int(_container_port), "protocol": _protocol} + port_map = { + "container_port": int(_container_port), + "protocol": _protocol + } if _host is None: result.append(port_map) - elif isinstance(_host, int) or isinstance(_host, str) and _host.isdigit(): + elif isinstance(_host, + int) or isinstance(_host, str) and _host.isdigit(): port_map["host_port"] = int(_host) result.append(port_map) elif isinstance(_host, tuple): @@ -556,16 +619,15 @@ def parse_host_port(_container_port, _protocol, _host): result.append(port_map) elif isinstance(_host, list): for host_list in _host: - host_list_result = parse_host_port(_container_port, _protocol, host_list) + host_list_result = parse_host_port(_container_port, + _protocol, host_list) result.extend(host_list_result) elif isinstance(_host, dict): _host_port = _host.get("port") if _host_port is not None: - if ( - isinstance(_host_port, int) - or isinstance(_host_port, str) - and _host_port.isdigit() - ): + if (isinstance(_host_port, int) + or isinstance(_host_port, str) + and _host_port.isdigit()): port_map["host_port"] = int(_host_port) elif isinstance(_host_port, tuple): port_map["host_ip"] = _host_port[0] @@ -588,10 +650,13 @@ def parse_host_port(_container_port, _protocol, _host): if "restart_policy" in args: params["restart_policy"] = args["restart_policy"].get("Name") - params["restart_tries"] = args["restart_policy"].get("MaximumRetryCount") + params["restart_tries"] = args["restart_policy"].get( + "MaximumRetryCount") args.pop("restart_policy") - params["resource_limits"]["pids"] = {"limit": args.pop("pids_limit", None)} + params["resource_limits"]["pids"] = { + "limit": args.pop("pids_limit", None) + } params["resource_limits"]["cpu"] = { "cpus": args.pop("cpuset_cpus", None), @@ -615,13 +680,11 @@ def parse_host_port(_container_port, _protocol, _host): } for item in args.pop("ulimits", []): - params["r_limits"].append( - { - "type": item["Name"], - "hard": item["Hard"], - "soft": item["Soft"], - } - ) + params["r_limits"].append({ + "type": item["Name"], + "hard": item["Hard"], + "soft": item["Soft"], + }) for item in args.pop("volumes", {}).items(): key, value = item @@ -674,8 +737,7 @@ def parse_host_port(_container_port, _protocol, _host): params["utsns"] = {"nsmode": args.pop("uts_mode")} if len(args) > 0: - raise TypeError( - "Unknown keyword argument(s): " + " ,".join(f"'{k}'" for k in args.keys()) - ) + raise TypeError("Unknown keyword argument(s): " + + " ,".join(f"'{k}'" for k in args.keys())) return params diff --git a/podman/domain/containers_manager.py b/podman/domain/containers_manager.py index 867ac424..adb2f654 100644 --- a/podman/domain/containers_manager.py +++ b/podman/domain/containers_manager.py @@ -26,7 +26,8 @@ def exists(self, key: str) -> bool: return response.ok def get(self, key: str) -> Container: - """Get container by name or id. + """ + Get container by name or id. Args: container_id: Container name or id. @@ -44,7 +45,8 @@ def get(self, key: str) -> Container: return self.prepare_model(attrs=response.json()) def list(self, **kwargs) -> List[Container]: - """Report on containers. + """ + Report on containers. Keyword Args: all: If False, only show running containers. Default: False. @@ -90,7 +92,8 @@ def list(self, **kwargs) -> List[Container]: return [self.prepare_model(attrs=i) for i in response.json()] def prune(self, filters: Mapping[str, str] = None) -> Dict[str, Any]: - """Delete stopped containers. + """ + Delete stopped containers. Args: filters: Criteria for determining containers to remove. Available keys are: @@ -115,7 +118,8 @@ def prune(self, filters: Mapping[str, str] = None) -> Dict[str, Any]: raise APIError( entry["Err"], response=response, - explanation=f"""Failed to prune container '{entry["Id"]}'""", + explanation= + f"""Failed to prune container '{entry["Id"]}'""", ) results["ContainersDeleted"].append(entry["Id"]) @@ -123,7 +127,8 @@ def prune(self, filters: Mapping[str, str] = None) -> Dict[str, Any]: return results def remove(self, container_id: Union[Container, str], **kwargs): - """Delete container. + """ + Delete container. Podman only @@ -143,5 +148,6 @@ def remove(self, container_id: Union[Container, str], **kwargs): "force": kwargs.get("force"), } - response = self.client.delete(f"/containers/{container_id}", params=params) + response = self.client.delete(f"/containers/{container_id}", + params=params) response.raise_for_status() diff --git a/podman/domain/containers_run.py b/podman/domain/containers_run.py index c88fc3fc..1924fcbb 100644 --- a/podman/domain/containers_run.py +++ b/podman/domain/containers_run.py @@ -75,7 +75,10 @@ def run( log_iter = None if log_type in ("json-file", "journald"): - log_iter = container.logs(stdout=stdout, stderr=stderr, stream=True, follow=True) + log_iter = container.logs(stdout=stdout, + stderr=stderr, + stream=True, + follow=True) exit_status = container.wait() if exit_status != 0: @@ -87,6 +90,8 @@ def run( container.remove() if exit_status != 0: - raise ContainerError(container, exit_status, command, image, log_iter) + raise ContainerError(container, exit_status, command, image, + log_iter) - return log_iter if kwargs.get("stream", False) or log_iter is None else b"".join(log_iter) + return log_iter if kwargs.get( + "stream", False) or log_iter is None else b"".join(log_iter) diff --git a/podman/domain/images.py b/podman/domain/images.py index 274873fe..b9d9b22e 100644 --- a/podman/domain/images.py +++ b/podman/domain/images.py @@ -46,7 +46,8 @@ def history(self) -> List[Dict[str, Any]]: def remove( self, **kwargs - ) -> List[Dict[api.Literal["Deleted", "Untagged", "Errors", "ExitCode"], Union[str, int]]]: + ) -> List[Dict[api.Literal["Deleted", "Untagged", "Errors", "ExitCode"], + Union[str, int]]]: """Delete image from Podman service. Podman only @@ -65,9 +66,9 @@ def remove( return self.manager.remove(self.id, **kwargs) def save( - self, - chunk_size: Optional[int] = api.DEFAULT_CHUNK_SIZE, - named: Union[str, bool] = False, # pylint: disable=unused-argument + self, + chunk_size: Optional[int] = api.DEFAULT_CHUNK_SIZE, + named: Union[str, bool] = False, # pylint: disable=unused-argument ) -> Iterator[bytes]: """Returns Image as tarball. @@ -81,17 +82,17 @@ def save( Raises: APIError: when service returns an error """ - response = self.client.get( - f"/images/{self.id}/get", params={"format": ["docker-archive"]}, stream=True - ) + response = self.client.get(f"/images/{self.id}/get", + params={"format": ["docker-archive"]}, + stream=True) response.raise_for_status(not_found=ImageNotFound) return response.iter_content(chunk_size=chunk_size) def tag( - self, - repository: str, - tag: Optional[str], - force: bool = False, # pylint: disable=unused-argument + self, + repository: str, + tag: Optional[str], + force: bool = False, # pylint: disable=unused-argument ) -> bool: """Tag Image into repository. diff --git a/podman/domain/images_build.py b/podman/domain/images_build.py index b25e951b..b9d0bb92 100644 --- a/podman/domain/images_build.py +++ b/podman/domain/images_build.py @@ -86,16 +86,18 @@ def build(self, **kwargs) -> Tuple[Image, Iterator[bytes]]: with open(filename, "w", encoding='utf-8') as file: shutil.copyfileobj(kwargs["fileobj"], file) - body = api.create_tar(anchor=path.name, gzip=kwargs.get("gzip", False)) + body = api.create_tar(anchor=path.name, + gzip=kwargs.get("gzip", False)) elif "path" in kwargs: filename = pathlib.Path(kwargs["path"]) / params["dockerfile"] # The Dockerfile will be copied into the context_dir if needed - params["dockerfile"] = api.prepare_containerfile(kwargs["path"], str(filename)) + params["dockerfile"] = api.prepare_containerfile( + kwargs["path"], str(filename)) excludes = api.prepare_containerignore(kwargs["path"]) - body = api.create_tar( - anchor=kwargs["path"], exclude=excludes, gzip=kwargs.get("gzip", False) - ) + body = api.create_tar(anchor=kwargs["path"], + exclude=excludes, + gzip=kwargs.get("gzip", False)) post_kwargs = {} if kwargs.get("timeout"): @@ -148,7 +150,8 @@ def _render_params(kwargs) -> Dict[str, List[Any]]: raise TypeError("Either path or fileobj must be provided.") if "gzip" in kwargs and "encoding" in kwargs: - raise PodmanError("Custom encoding not supported when gzip enabled.") + raise PodmanError( + "Custom encoding not supported when gzip enabled.") params = { "dockerfile": kwargs.get("dockerfile"), @@ -189,7 +192,8 @@ def _render_params(kwargs) -> Dict[str, List[Any]]: params["labels"] = json.dumps(kwargs.get("labels")) if params["dockerfile"] is None: - params["dockerfile"] = f".containerfile.{random.getrandbits(160):x}" + params[ + "dockerfile"] = f".containerfile.{random.getrandbits(160):x}" # Remove any unset parameters return dict(filter(lambda i: i[1] is not None, params.items())) diff --git a/podman/domain/images_manager.py b/podman/domain/images_manager.py index 13b2eed1..8f3cc3d8 100644 --- a/podman/domain/images_manager.py +++ b/podman/domain/images_manager.py @@ -78,9 +78,9 @@ def get(self, name: str) -> Image: # pylint: disable=arguments-differ,arguments return self.prepare_model(response.json()) def get_registry_data( - self, - name: str, - auth_config=Mapping[str, str], # pylint: disable=unused-argument + self, + name: str, + auth_config=Mapping[str, str], # pylint: disable=unused-argument ) -> RegistryData: """Returns registry data for an image. @@ -115,8 +115,9 @@ def load(self, data: bytes) -> Generator[Image, None, None]: # headers = {"Content-type": "application/x-www-form-urlencoded"} response = self.client.post( - "/images/load", data=data, headers={"Content-type": "application/x-tar"} - ) + "/images/load", + data=data, + headers={"Content-type": "application/x-tar"}) response.raise_for_status() body = response.json() @@ -124,7 +125,8 @@ def load(self, data: bytes) -> Generator[Image, None, None]: yield self.get(item) def prune( - self, filters: Optional[Mapping[str, Any]] = None + self, + filters: Optional[Mapping[str, Any]] = None ) -> Dict[Literal["ImagesDeleted", "SpaceReclaimed"], Any]: """Delete unused images. @@ -140,8 +142,7 @@ def prune( APIError: when service returns an error """ response = self.client.post( - "/images/prune", params={"filters": api.prepare_filters(filters)} - ) + "/images/prune", params={"filters": api.prepare_filters(filters)}) response.raise_for_status() deleted: List[Dict[str, str]] = [] @@ -152,21 +153,22 @@ def prune( error.append(element["Err"]) else: reclaimed += element["Size"] - deleted.append( - { - "Deleted": element["Id"], - "Untagged": "", - } - ) + deleted.append({ + "Deleted": element["Id"], + "Untagged": "", + }) if len(error) > 0: - raise APIError(response.url, response=response, explanation="; ".join(error)) + raise APIError(response.url, + response=response, + explanation="; ".join(error)) return { "ImagesDeleted": deleted, "SpaceReclaimed": reclaimed, } - def prune_builds(self) -> Dict[Literal["CachesDeleted", "SpaceReclaimed"], Any]: + def prune_builds( + self) -> Dict[Literal["CachesDeleted", "SpaceReclaimed"], Any]: """Delete builder cache. Method included to complete API, the operation always returns empty @@ -174,9 +176,10 @@ def prune_builds(self) -> Dict[Literal["CachesDeleted", "SpaceReclaimed"], Any]: """ return {"CachesDeleted": [], "SpaceReclaimed": 0} - def push( - self, repository: str, tag: Optional[str] = None, **kwargs - ) -> Union[str, Iterator[Union[str, Dict[str, Any]]]]: + def push(self, + repository: str, + tag: Optional[str] = None, + **kwargs) -> Union[str, Iterator[Union[str, Dict[str, Any]]]]: """Push Image or repository to the registry. Args: @@ -198,7 +201,8 @@ def push( headers = { # A base64url-encoded auth configuration - "X-Registry-Auth": encode_auth_header(auth_config) if auth_config else "" + "X-Registry-Auth": + encode_auth_header(auth_config) if auth_config else "" } params = { @@ -208,13 +212,16 @@ def push( name = f'{repository}:{tag}' if tag else repository name = urllib.parse.quote_plus(name) - response = self.client.post(f"/images/{name}/push", params=params, headers=headers) + response = self.client.post(f"/images/{name}/push", + params=params, + headers=headers) response.raise_for_status(not_found=ImageNotFound) tag_count = 0 if tag is None else 1 body = [ { - "status": f"Pushing repository {repository} ({tag_count} tags)", + "status": + f"Pushing repository {repository} ({tag_count} tags)", }, { "status": "Pushing", @@ -235,8 +242,9 @@ def push( @staticmethod def _push_helper( - decode: bool, body: List[Dict[str, Any]] - ) -> Iterator[Union[str, Dict[str, Any]]]: + decode: bool, + body: List[Dict[str, + Any]]) -> Iterator[Union[str, Dict[str, Any]]]: """Helper needed to allow push() to return either a generator or a str.""" for entry in body: if decode: @@ -245,9 +253,11 @@ def _push_helper( yield json.dumps(entry) # pylint: disable=too-many-locals,too-many-branches - def pull( - self, repository: str, tag: Optional[str] = None, all_tags: bool = False, **kwargs - ) -> Union[Image, List[Image], Iterator[str]]: + def pull(self, + repository: str, + tag: Optional[str] = None, + all_tags: bool = False, + **kwargs) -> Union[Image, List[Image], Iterator[str]]: """Request Podman service to pull image(s) from repository. Args: @@ -283,7 +293,8 @@ def pull( headers = { # A base64url-encoded auth configuration - "X-Registry-Auth": encode_auth_header(auth_config) if auth_config else "" + "X-Registry-Auth": + encode_auth_header(auth_config) if auth_config else "" } params = { @@ -299,7 +310,8 @@ def pull( if "platform" in kwargs: tokens = kwargs.get("platform").split("/") if 1 < len(tokens) > 3: - raise ValueError(f'\'{kwargs.get("platform")}\' is not a legal platform.') + raise ValueError( + f'\'{kwargs.get("platform")}\' is not a legal platform.') params["OS"] = tokens[0] if len(tokens) > 1: @@ -308,7 +320,10 @@ def pull( params["Variant"] = tokens[2] stream = kwargs.get("stream", False) - response = self.client.post("/images/pull", params=params, stream=stream, headers=headers) + response = self.client.post("/images/pull", + params=params, + stream=stream, + headers=headers) response.raise_for_status(not_found=ImageNotFound) if stream: @@ -331,7 +346,8 @@ def remove( image: Union[Image, str], force: Optional[bool] = None, noprune: bool = False, # pylint: disable=unused-argument - ) -> List[Dict[Literal["Deleted", "Untagged", "Errors", "ExitCode"], Union[str, int]]]: + ) -> List[Dict[Literal["Deleted", "Untagged", "Errors", "ExitCode"], Union[ + str, int]]]: """Delete image from Podman service. Args: @@ -346,7 +362,8 @@ def remove( if isinstance(image, Image): image = image.id - response = self.client.delete(f"/images/{image}", params={"force": force}) + response = self.client.delete(f"/images/{image}", + params={"force": force}) response.raise_for_status(not_found=ImageNotFound) body = response.json() diff --git a/podman/domain/ipam.py b/podman/domain/ipam.py index 25012513..9f844d17 100644 --- a/podman/domain/ipam.py +++ b/podman/domain/ipam.py @@ -24,14 +24,12 @@ def __init__( aux_addresses: Ignored. """ super().__init__() - self.update( - { - "AuxiliaryAddresses": aux_addresses, - "Gateway": gateway, - "IPRange": iprange, - "Subnet": subnet, - } - ) + self.update({ + "AuxiliaryAddresses": aux_addresses, + "Gateway": gateway, + "IPRange": iprange, + "Subnet": subnet, + }) class IPAMConfig(dict): @@ -51,10 +49,8 @@ def __init__( options: Options to provide to the Network driver. """ super().__init__() - self.update( - { - "Config": pool_configs or [], - "Driver": driver, - "Options": options or {}, - } - ) + self.update({ + "Config": pool_configs or [], + "Driver": driver, + "Options": options or {}, + }) diff --git a/podman/domain/manager.py b/podman/domain/manager.py index 6ba4faea..25236015 100644 --- a/podman/domain/manager.py +++ b/podman/domain/manager.py @@ -6,7 +6,8 @@ from podman.api.client import APIClient # Methods use this Type when a subclass of PodmanResource is expected. -PodmanResourceType: TypeVar = TypeVar("PodmanResourceType", bound="PodmanResource") +PodmanResourceType: TypeVar = TypeVar("PodmanResourceType", + bound="PodmanResource") class PodmanResource(ABC): @@ -103,7 +104,9 @@ def get(self, key: str) -> PodmanResourceType: def list(self, **kwargs) -> List[PodmanResourceType]: """Returns list of resources.""" - def prepare_model(self, attrs: Union[PodmanResource, Mapping[str, Any]]) -> PodmanResourceType: + def prepare_model( + self, attrs: Union[PodmanResource, + Mapping[str, Any]]) -> PodmanResourceType: """Create a model from a set of attributes.""" # Refresh existing PodmanResource. @@ -116,7 +119,9 @@ def prepare_model(self, attrs: Union[PodmanResource, Mapping[str, Any]]) -> Podm if isinstance(attrs, abc.Mapping): # TODO Determine why pylint is reporting typing.Type not callable # pylint: disable=not-callable - return self.resource(attrs=attrs, client=self.client, collection=self) + return self.resource(attrs=attrs, + client=self.client, + collection=self) # pylint: disable=broad-exception-raised raise Exception(f"Can't create {self.resource.__name__} from {attrs}") diff --git a/podman/domain/manifests.py b/podman/domain/manifests.py index bff805ff..51ee6632 100644 --- a/podman/domain/manifests.py +++ b/podman/domain/manifests.py @@ -91,9 +91,9 @@ def add(self, images: List[Union[Image, str]], **kwargs) -> None: return self.reload() def push( - self, - destination: str, - all: Optional[bool] = None, # pylint: disable=redefined-builtin + self, + destination: str, + all: Optional[bool] = None, # pylint: disable=redefined-builtin ) -> None: """Push a manifest list or image index to a registry. @@ -109,7 +109,8 @@ def push( "all": all, "destination": destination, } - response = self.client.post(f"/manifests/{self.quoted_name}/push", params=params) + response = self.client.post(f"/manifests/{self.quoted_name}/push", + params=params) response.raise_for_status() def remove(self, digest: str) -> None: @@ -148,10 +149,10 @@ def resource(self): return Manifest def create( - self, - name: str, - images: Optional[List[Union[Image, str]]] = None, - all: Optional[bool] = None, # pylint: disable=redefined-builtin + self, + name: str, + images: Optional[List[Union[Image, str]]] = None, + all: Optional[bool] = None, # pylint: disable=redefined-builtin ) -> Manifest: """Create a Manifest. @@ -217,7 +218,8 @@ def get(self, key: str) -> Manifest: def list(self, **kwargs) -> List[Manifest]: """Not Implemented.""" - raise NotImplementedError("Podman service currently does not support listing manifests.") + raise NotImplementedError( + "Podman service currently does not support listing manifests.") def remove(self, name: Union[Manifest, str]) -> Dict[str, Any]: """Delete the manifest list from the Podman service.""" diff --git a/podman/domain/networks.py b/podman/domain/networks.py index e1af5ba4..d90cf33c 100644 --- a/podman/domain/networks.py +++ b/podman/domain/networks.py @@ -43,7 +43,10 @@ def containers(self): """List[Container]: Returns list of Containers connected to network.""" with suppress(KeyError): container_manager = ContainersManager(client=self.client) - return [container_manager.get(ident) for ident in self.attrs["Containers"].keys()] + return [ + container_manager.get(ident) + for ident in self.attrs["Containers"].keys() + ] return [] @property @@ -89,22 +92,31 @@ def connect(self, container: Union[str, Container], *_, **kwargs) -> None: "IPv6Address": kwargs.get('ipv6_address'), "Links": kwargs.get("link_local_ips"), } - ipam = {k: v for (k, v) in ipam.items() if not (v is None or len(v) == 0)} + ipam = { + k: v + for (k, v) in ipam.items() if not (v is None or len(v) == 0) + } endpoint_config = { "Aliases": kwargs.get("aliases"), "DriverOpts": kwargs.get("driver_opt"), - "IPAddress": kwargs.get("ipv4_address", kwargs.get("ipv6_address")), + "IPAddress": kwargs.get("ipv4_address", + kwargs.get("ipv6_address")), "IPAMConfig": ipam, "Links": kwargs.get("link_local_ips"), "NetworkID": self.id, } endpoint_config = { - k: v for (k, v) in endpoint_config.items() if not (v is None or len(v) == 0) + k: v + for (k, v) in endpoint_config.items() + if not (v is None or len(v) == 0) } data = {"Container": container, "EndpointConfig": endpoint_config} - data = {k: v for (k, v) in data.items() if not (v is None or len(v) == 0)} + data = { + k: v + for (k, v) in data.items() if not (v is None or len(v) == 0) + } response = self.client.post( f"/networks/{self.name}/connect", @@ -129,7 +141,8 @@ def disconnect(self, container: Union[str, Container], **kwargs) -> None: container = container.id data = {"Container": container, "Force": kwargs.get("force")} - response = self.client.post(f"/networks/{self.name}/disconnect", data=json.dumps(data)) + response = self.client.post(f"/networks/{self.name}/disconnect", + data=json.dumps(data)) response.raise_for_status() def remove(self, force: Optional[bool] = None, **kwargs) -> None: diff --git a/podman/domain/networks_manager.py b/podman/domain/networks_manager.py index 3d5034be..549c275e 100644 --- a/podman/domain/networks_manager.py +++ b/podman/domain/networks_manager.py @@ -157,7 +157,8 @@ def list(self, **kwargs) -> List[Network]: return [self.prepare_model(i) for i in response.json()] def prune( - self, filters: Optional[Dict[str, Any]] = None + self, + filters: Optional[Dict[str, Any]] = None ) -> Dict[api.Literal["NetworksDeleted", "SpaceReclaimed"], Any]: """Delete unused Networks. @@ -185,7 +186,9 @@ def prune( return {"NetworksDeleted": deleted, "SpaceReclaimed": 0} - def remove(self, name: [Network, str], force: Optional[bool] = None) -> None: + def remove(self, + name: [Network, str], + force: Optional[bool] = None) -> None: """Remove Network resource. Args: @@ -198,5 +201,6 @@ def remove(self, name: [Network, str], force: Optional[bool] = None) -> None: if isinstance(name, Network): name = name.name - response = self.client.delete(f"/networks/{name}", params={"force": force}) + response = self.client.delete(f"/networks/{name}", + params={"force": force}) response.raise_for_status() diff --git a/podman/domain/pods.py b/podman/domain/pods.py index f2a72ec8..4a28177b 100644 --- a/podman/domain/pods.py +++ b/podman/domain/pods.py @@ -31,7 +31,8 @@ def kill(self, signal: Union[str, int, None] = None) -> None: NotFound: when pod not found APIError: when service reports an error """ - response = self.client.post(f"/pods/{self.id}/kill", params={"signal": signal}) + response = self.client.post(f"/pods/{self.id}/kill", + params={"signal": signal}) response.raise_for_status() def pause(self) -> None: diff --git a/podman/domain/pods_manager.py b/podman/domain/pods_manager.py index 826f898d..5d9a8c1f 100644 --- a/podman/domain/pods_manager.py +++ b/podman/domain/pods_manager.py @@ -83,7 +83,8 @@ def list(self, **kwargs) -> List[Pod]: response.raise_for_status() return [self.prepare_model(attrs=i) for i in response.json()] - def prune(self, filters: Optional[Dict[str, str]] = None) -> Dict[str, Any]: + def prune(self, + filters: Optional[Dict[str, str]] = None) -> Dict[str, Any]: """Delete unused Pods. Returns: @@ -94,7 +95,8 @@ def prune(self, filters: Optional[Dict[str, str]] = None) -> Dict[str, Any]: Raises: APIError: when service reports error """ - response = self.client.post("/pods/prune", params={"filters": api.prepare_filters(filters)}) + response = self.client.post( + "/pods/prune", params={"filters": api.prepare_filters(filters)}) response.raise_for_status() deleted: List[str] = [] @@ -108,7 +110,9 @@ def prune(self, filters: Optional[Dict[str, str]] = None) -> Dict[str, Any]: deleted.append(item["Id"]) return {"PodsDeleted": deleted, "SpaceReclaimed": 0} - def remove(self, pod_id: Union[Pod, str], force: Optional[bool] = None) -> None: + def remove(self, + pod_id: Union[Pod, str], + force: Optional[bool] = None) -> None: """Delete pod. Args: @@ -125,10 +129,13 @@ def remove(self, pod_id: Union[Pod, str], force: Optional[bool] = None) -> None: if isinstance(pod_id, Pod): pod_id = pod_id.id - response = self.client.delete(f"/pods/{pod_id}", params={"force": force}) + response = self.client.delete(f"/pods/{pod_id}", + params={"force": force}) response.raise_for_status() - def stats(self, **kwargs) -> Union[List[Dict[str, Any]], Iterator[List[Dict[str, Any]]]]: + def stats( + self, **kwargs + ) -> Union[List[Dict[str, Any]], Iterator[List[Dict[str, Any]]]]: """Resource usage statistics for the containers in pods. Keyword Args: @@ -142,7 +149,8 @@ def stats(self, **kwargs) -> Union[List[Dict[str, Any]], Iterator[List[Dict[str, APIError: when service reports an error """ if "all" in kwargs and "name" in kwargs: - raise ValueError("Keywords 'all' and 'name' are mutually exclusive.") + raise ValueError( + "Keywords 'all' and 'name' are mutually exclusive.") # Keeping the default for stream as False to not break existing users # Should probably be changed in a newer major version to match behavior of container.stats diff --git a/podman/domain/registry_data.py b/podman/domain/registry_data.py index a970dcec..20f9face 100644 --- a/podman/domain/registry_data.py +++ b/podman/domain/registry_data.py @@ -56,7 +56,8 @@ def has_platform(self, platform: Union[str, Mapping[str, Any]]) -> bool: InvalidArgument: when platform value is not valid APIError: when service reports an error """ - invalid_platform = InvalidArgument(f"'{platform}' is not a valid platform descriptor.") + invalid_platform = InvalidArgument( + f"'{platform}' is not a valid platform descriptor.") if platform is None: platform = {} @@ -65,7 +66,8 @@ def has_platform(self, platform: Union[str, Mapping[str, Any]]) -> bool: if not {"os", "architecture"} <= platform.keys(): version = self.client.version() platform["os"] = platform.get("os", version["Os"]) - platform["architecture"] = platform.get("architecture", version["Arch"]) + platform["architecture"] = platform.get( + "architecture", version["Arch"]) elif isinstance(platform, str): elements = platform.split("/") if 1 < len(elements) > 3: @@ -82,5 +84,4 @@ def has_platform(self, platform: Union[str, Mapping[str, Any]]) -> bool: return ( # Variant not carried in libpod attrs platform["os"] == self.attrs["Os"] - and platform["architecture"] == self.attrs["Architecture"] - ) + and platform["architecture"] == self.attrs["Architecture"]) diff --git a/podman/domain/secrets.py b/podman/domain/secrets.py index f89e25c3..4c297da3 100644 --- a/podman/domain/secrets.py +++ b/podman/domain/secrets.py @@ -24,8 +24,8 @@ def name(self): return "" def remove( - self, - all: Optional[bool] = None, # pylint: disable=redefined-builtin + self, + all: Optional[bool] = None, # pylint: disable=redefined-builtin ): """Delete secret. @@ -109,16 +109,18 @@ def create( "name": name, "driver": driver, } - response = self.client.post("/secrets/create", params=params, data=data) + response = self.client.post("/secrets/create", + params=params, + data=data) response.raise_for_status() body = response.json() return self.get(body["ID"]) def remove( - self, - secret_id: Union[Secret, str], - all: Optional[bool] = None, # pylint: disable=redefined-builtin + self, + secret_id: Union[Secret, str], + all: Optional[bool] = None, # pylint: disable=redefined-builtin ): """Delete secret. @@ -135,5 +137,6 @@ def remove( if isinstance(secret_id, Secret): secret_id = secret_id.id - response = self.client.delete(f"/secrets/{secret_id}", params={"all": all}) + response = self.client.delete(f"/secrets/{secret_id}", + params={"all": all}) response.raise_for_status() diff --git a/podman/domain/system.py b/podman/domain/system.py index c4a8794d..719f9f01 100644 --- a/podman/domain/system.py +++ b/podman/domain/system.py @@ -36,13 +36,13 @@ def info(self, *_, **__) -> Dict[str, Any]: return response.json() def login( - self, - username: str, - password: Optional[str] = None, - email: Optional[str] = None, - registry: Optional[str] = None, - reauth: Optional[bool] = False, # pylint: disable=unused-argument - dockercfg_path: Optional[str] = None, # pylint: disable=unused-argument + self, + username: str, + password: Optional[str] = None, + email: Optional[str] = None, + registry: Optional[str] = None, + reauth: Optional[bool] = False, # pylint: disable=unused-argument + dockercfg_path: Optional[str] = None, # pylint: disable=unused-argument ) -> Dict[str, Any]: """Log into Podman service. diff --git a/podman/domain/volumes.py b/podman/domain/volumes.py index e1377c9b..3e030726 100644 --- a/podman/domain/volumes.py +++ b/podman/domain/volumes.py @@ -102,7 +102,8 @@ def list(self, *_, **kwargs) -> List[Volume]: - name (str): filter by volume's name """ filters = api.prepare_filters(kwargs.get("filters")) - response = self.client.get("/volumes/json", params={"filters": filters}) + response = self.client.get("/volumes/json", + params={"filters": filters}) if response.status_code == requests.codes.not_found: return [] @@ -111,7 +112,8 @@ def list(self, *_, **kwargs) -> List[Volume]: return [self.prepare_model(i) for i in response.json()] def prune( - self, filters: Optional[Dict[str, str]] = None # pylint: disable=unused-argument + self, + filters: Optional[Dict[str, str]] = None # pylint: disable=unused-argument ) -> Dict[Literal["VolumesDeleted", "SpaceReclaimed"], Any]: """Delete unused volumes. @@ -132,14 +134,17 @@ def prune( raise APIError( item["Err"], response=response, - explanation=f"""Failed to prune volume '{item.get("Id")}'""", + explanation= + f"""Failed to prune volume '{item.get("Id")}'""", ) volumes.append(item.get("Id")) space_reclaimed += item["Size"] return {"VolumesDeleted": volumes, "SpaceReclaimed": space_reclaimed} - def remove(self, name: Union[Volume, str], force: Optional[bool] = None) -> None: + def remove(self, + name: Union[Volume, str], + force: Optional[bool] = None) -> None: """Delete a volume. Podman only. @@ -153,5 +158,6 @@ def remove(self, name: Union[Volume, str], force: Optional[bool] = None) -> None """ if isinstance(name, Volume): name = name.name - response = self.client.delete(f"/volumes/{name}", params={"force": force}) + response = self.client.delete(f"/volumes/{name}", + params={"force": force}) response.raise_for_status()