from dotenv import load_dotenv from flask import Flask, render_template from flask_apscheduler import APScheduler from flask_login import LoginManager import docker import time import os #DEBUG from pprint import pprint def calculate_cpu_percent(stats, status): if (status != "running"): return 0 cpu_count = len(stats["cpu_stats"]["cpu_usage"]["percpu_usage"]) cpu_percent = 0.0 cpu_delta = float(stats["cpu_stats"]["cpu_usage"]["total_usage"]) - \ float(stats["precpu_stats"]["cpu_usage"]["total_usage"]) system_delta = float(stats["cpu_stats"]["system_cpu_usage"]) - \ float(stats["precpu_stats"]["system_cpu_usage"]) if system_delta > 0.0: cpu_percent = cpu_delta / system_delta * 100.0 * cpu_count return round(cpu_percent, 2) def calculate_ram_percent(stats, status): if (status != "running"): return 0 mem_used = stats["memory_stats"]["usage"] - stats["memory_stats"]["stats"]["cache"] + stats["memory_stats"]["stats"]["active_file"] limit = stats['memory_stats']['limit'] return round(mem_used / limit * 100, 2) client = docker.from_env() #login_manager = LoginManager() app = Flask(__name__) scheduler = APScheduler() load_dotenv() #login_manager.init_app(app) #https://flask-login.readthedocs.io/en/latest/ @app.route("/", methods=['GET']) def index(): containers = client.containers.list(all=True) return render_template("index.html", title="test", containers_len=len(containers), containers=containers) @app.route("/parts/modal/container/", methods=['GET']) def parts_container_modal(container_id): container = client.containers.get(container_id) stats = container.stats(stream=False) #com.docker.extension.detailed-description project = '' if('com.docker.compose.project' in container.attrs['Config']['Labels']): project = container.attrs['Config']['Labels']['com.docker.compose.project'] return render_template("container_modal.html", name=container.name, project=project, ports=container.attrs['NetworkSettings']['Ports'], status=container.status, CPU=calculate_cpu_percent(stats, container.status) ,RAM =calculate_ram_percent(stats, container.status) ) @app.route("/api//", methods=['DELETE', 'POST']) def api_stop_container(container_id, action): container = client.containers.get(container_id) #pprint(vars(container)) try: match action: case 'stop': container.stop() case 'start': container.start() case _: return 'False' container = client.containers.get(container_id) return container.status except: return 'False' def scheduleTasks(): print("This test runs every 3 seconds") if __name__ == '__main__': #scheduler.add_job(id = 'scheduled_tasks', func=scheduleTasks, trigger="interval", seconds=3) #scheduler.start() app.run(host=os.getenv('IP'), port=os.getenv('port'), debug=True)