1+ import argparse
2+ import json
3+ import logging
4+ import sys
5+ import time
6+ from typing import Any
7+ from docker import DockerClient , from_env
8+ from docker .models .images import Image
9+ from docker .errors import APIError
10+
11+ def read_config_json (path : str ) -> dict [str , Any ]:
12+ with open (path , "r" , encoding = 'utf-8' ) as file :
13+ data = json .load (file )
14+ return data
15+
16+ def log_level_by_str (level : str ) -> int :
17+ ll = level .upper ()
18+
19+ if ll == "DEBUG" :
20+ return logging .DEBUG
21+ elif ll == "INFO" :
22+ return logging .INFO
23+ elif ll == "WARNING" :
24+ return logging .WARNING
25+ elif ll == "ERROR" :
26+ return logging .ERROR
27+ elif ll == "FATAL" :
28+ return logging .FATAL
29+ else :
30+ raise Exception (f"Unknown log level '{ level } '" )
31+
32+ class __DockerContainers :
33+ def __init__ (self , client : DockerClient ) -> None :
34+ self .__client : DockerClient = client
35+
36+ def get_list (self ):
37+ return self .__client .containers .list ()
38+
39+ class __DockerImages :
40+ def __init__ (self , client : DockerClient ) -> None :
41+ self .__client : DockerClient = client
42+
43+ def get_image_info (self , image : Image ) -> dict [str , str ]:
44+ return {
45+ "label" : image .tags [0 ],
46+ "id" : str (image .id ).replace ("sha256:" , "" , 1 ),
47+ "short_id" : image .short_id .replace ("sha256:" , "" , 1 )
48+ }
49+
50+ def get_list (
51+ self ,
52+ name : str | None = None ,
53+ all : bool = False ,
54+ filters : dict [str , Any ] | None = None
55+ ):
56+ return self .__client .images .list (name = name , all = all , filters = filters )
57+
58+ def remove (
59+ self ,
60+ images : list [Image ],
61+ containers : list ,
62+ white_list : list [str ] = [],
63+ force : bool = False ,
64+ noprune : bool = False
65+ ) -> list [Image ]:
66+ deleted_images : list [Image ] = []
67+ used_images = {container .image .id for container in containers }
68+
69+ for img in images :
70+ img_info = self .get_image_info (img )
71+
72+ if (
73+ img .id in used_images
74+ or img_info ["id" ] in white_list
75+ or img_info ["short_id" ] in white_list
76+ or img_info ["label" ] in white_list
77+ ):
78+ continue
79+
80+ try :
81+ self .__client .images .remove (
82+ image = img_info ["id" ],
83+ force = force ,
84+ noprune = noprune
85+ )
86+ deleted_images .append (img )
87+ except APIError : # Triggers when the image is in use.
88+ pass
89+ except Exception as e :
90+ logging .error (f"An error occurred while removing image '{ img } ': { e } " )
91+
92+ return deleted_images
93+
94+ if __name__ == '__main__' :
95+ parser = argparse .ArgumentParser ()
96+ parser .add_argument ('--config' , type = str , required = True , help = 'Path to configuration' )
97+ args = parser .parse_args ()
98+
99+ cfg = read_config_json (args .config )
100+ cfg_this = cfg ["remove_unused_images" ]
101+
102+ logging .basicConfig (
103+ stream = sys .stderr ,
104+ level = log_level_by_str (cfg ["log_level" ]),
105+ datefmt = '%Y-%m-%dT%H:%M:%S%z' ,
106+ format = '%(asctime)s - %(levelname)s - %(message)s'
107+ )
108+
109+ client = from_env ()
110+
111+ while True :
112+ images = __DockerImages (client )
113+ containers = __DockerContainers (client )
114+
115+ for i in images .remove (images = images .get_list (all = True ), containers = containers .get_list (), white_list = cfg_this ["white_list" ]):
116+ logging .info (f"Deleted images: { images .get_image_info (i )} " )
117+
118+ time .sleep (cfg_this ["per_seconds" ])
0 commit comments