Spaces:
Paused
Paused
File size: 4,840 Bytes
deab47c 43eeb15 deab47c edbeadb 20bc5ae deab47c d6e60a2 1e2ffc3 deab47c edbeadb d6e60a2 edbeadb d6e60a2 edbeadb deab47c d6e60a2 deab47c d6e60a2 deab47c edbeadb deab47c 9138bc4 deab47c 9138bc4 d6e60a2 20bc5ae 9138bc4 875ddcf 9138bc4 d6e60a2 deab47c d6e60a2 20bc5ae deab47c 20bc5ae deab47c 20bc5ae d6e60a2 2359298 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
import time
import docker
import atexit
from typing import Optional
from python.helpers.files import get_abs_path
from python.helpers.errors import format_error
from python.helpers.print_style import PrintStyle
from python.helpers.log import Log
class DockerContainerManager:
def __init__(self, image: str, name: str, ports: Optional[dict[str, int]] = None, volumes: Optional[dict[str, dict[str, str]]] = None,logger: Log|None=None):
self.logger = logger
self.image = image
self.name = name
self.ports = ports
self.volumes = volumes
self.init_docker()
def init_docker(self):
self.client = None
while not self.client:
try:
self.client = docker.from_env()
self.container = None
except Exception as e:
err = format_error(e)
if ("ConnectionRefusedError(61," in err or "Error while fetching server API version" in err):
PrintStyle.hint("Connection to Docker failed. Is docker or Docker Desktop running?") # hint for user
if self.logger:self.logger.log(type="hint", content="Connection to Docker failed. Is docker or Docker Desktop running?")
PrintStyle.error(err)
if self.logger:self.logger.log(type="error", content=err)
time.sleep(5) # try again in 5 seconds
else: raise
return self.client
def cleanup_container(self) -> None:
if self.container:
try:
self.container.stop()
self.container.remove()
PrintStyle.standard(f"Stopped and removed the container: {self.container.id}")
if self.logger: self.logger.log(type="info", content=f"Stopped and removed the container: {self.container.id}")
except Exception as e:
PrintStyle.error(f"Failed to stop and remove the container: {e}")
if self.logger: self.logger.log(type="error", content=f"Failed to stop and remove the container: {e}")
def get_image_containers(self):
if not self.client: self.client = self.init_docker()
containers = self.client.containers.list(all=True, filters={"ancestor": self.image})
infos = []
for container in containers:
infos.append({
"id": container.id,
"name": container.name,
"status": container.status,
"image": container.image,
"ports": container.ports,
"web_port": (container.ports.get("80/tcp") or [{}])[0].get("HostPort"),
"ssh_port": (container.ports.get("22/tcp") or [{}])[0].get("HostPort"),
# "volumes": container.volumes,
# "data_folder": container.volumes["/a0"],
})
return infos
def start_container(self) -> None:
if not self.client: self.client = self.init_docker()
existing_container = None
for container in self.client.containers.list(all=True):
if container.name == self.name:
existing_container = container
break
if existing_container:
if existing_container.status != 'running':
PrintStyle.standard(f"Starting existing container: {self.name} for safe code execution...")
if self.logger: self.logger.log(type="info", content=f"Starting existing container: {self.name} for safe code execution...", temp=True)
existing_container.start()
self.container = existing_container
time.sleep(2) # this helps to get SSH ready
else:
self.container = existing_container
# PrintStyle.standard(f"Container with name '{self.name}' is already running with ID: {existing_container.id}")
else:
PrintStyle.standard(f"Initializing docker container {self.name} for safe code execution...")
if self.logger: self.logger.log(type="info", content=f"Initializing docker container {self.name} for safe code execution...", temp=True)
self.container = self.client.containers.run(
self.image,
detach=True,
ports=self.ports, # type: ignore
name=self.name,
volumes=self.volumes, # type: ignore
)
# atexit.register(self.cleanup_container)
PrintStyle.standard(f"Started container with ID: {self.container.id}")
if self.logger: self.logger.log(type="info", content=f"Started container with ID: {self.container.id}")
time.sleep(5) # this helps to get SSH ready
|