Pratyush Desai
87feb8fda6
switched port field to a JSONField adds a helper function to display server port mapping in admin launching a container will map the ports appropriately Instead of typing json to map ports we have a widget replaced by such text fields. where one can + button to add or - to remove i.e. [container_port/proto]:[host_port] At first port was an IntegerField. Hence i added a manual migration which is the latest one to get them to be default:port(old value) Signed-off-by: Pratyush Desai <pratyush.desai@liberta.casa>
129 lines
4.9 KiB
Python
129 lines
4.9 KiB
Python
from django.db import models
|
|
import podman
|
|
import shlex
|
|
import re
|
|
from typing import Optional,List
|
|
|
|
class Game(models.Model):
|
|
name = models.CharField(max_length=100)
|
|
genre = models.CharField(max_length=50, blank=True, null=True)
|
|
thumbnail = models.ImageField(upload_to='game_thumbnails/', null=True, blank=True)
|
|
|
|
def __str__(self) -> str:
|
|
return self.name
|
|
|
|
class Server(models.Model):
|
|
STATUS_CHOICES = [
|
|
('online', 'Online'),
|
|
('offline', 'Offline'),
|
|
]
|
|
last_log = models.TextField(blank=True, null=True)
|
|
game = models.ForeignKey(Game, on_delete=models.CASCADE)
|
|
name = models.CharField(max_length=100)
|
|
ip_address = models.GenericIPAddressField(null=True)
|
|
port = models.JSONField(default=dict, blank=True)
|
|
image = models.CharField(max_length=200, null=True)
|
|
run_command = models.CharField(max_length=500, blank=True, null=True)
|
|
command_args = models.TextField(blank=True, null=True)
|
|
container_id = models.CharField(max_length=64, blank=True, null=True)
|
|
status = models.CharField(max_length=10, choices=STATUS_CHOICES, default='offline')
|
|
|
|
def __str__(self)-> str:
|
|
return f"{self.game.name} Server at {self.ip_address}:{self.port}"
|
|
|
|
@property
|
|
def safe_name(self) -> str:
|
|
"""Return a container-safe version of the server name"""
|
|
return re.sub(r'[^a-zA-Z0-9_.-]', '_', self.name)
|
|
|
|
def _log_error(self, msg):
|
|
self.last_log = msg
|
|
self.save(update_fields=["status", "last_log"])
|
|
def _get_podman_client(self) -> podman.PodmanClient:
|
|
"""Get a configured Podman client instance."""
|
|
return podman.PodmanClient(base_url="unix:///run/user/1000/podman/podman.sock")
|
|
|
|
def sync_status(self) -> None:
|
|
"""Check the real-time status of the container and update the field."""
|
|
try:
|
|
container = self._get_podman_client().containers.get(self.safe_name)
|
|
if container.status == "running":
|
|
self.status = "online"
|
|
else:
|
|
self.status = "offline"
|
|
except podman.errors.NotFound:
|
|
self.status = "offline"
|
|
except Exception as e:
|
|
self.status = "offline"
|
|
self.save()
|
|
|
|
def get_ports_display(self):
|
|
return ", ".join(f"{k}→{v}" for k, v in self.port.items())
|
|
get_ports_display.short_description = "Ports" # display name in admin
|
|
|
|
|
|
def launch_pod_container(self) -> str:
|
|
try:
|
|
port_bindings = {
|
|
container_port: ('0.0.0.0', host_port)
|
|
for container_port, host_port in self.port.items()
|
|
}
|
|
|
|
container = self._get_podman_client().containers.create(
|
|
name=self.safe_name,
|
|
image=self.image,
|
|
ports=port_bindings,
|
|
command=shlex.split(self.run_command),
|
|
detach=True,
|
|
)
|
|
container.start()
|
|
self.container_id = container.id
|
|
self.last_log = f"Launched container {container.id}"
|
|
self.sync_status()
|
|
self.save()
|
|
return f"Container launched successfully: {container.id}"
|
|
except podman.errors.APIError as e:
|
|
self.last_log = f"API Error: {str(e)}"
|
|
self.save()
|
|
return f"API Error: {e}"
|
|
except Exception as e:
|
|
self.last_log = f"Error: {str(e)}"
|
|
self.save()
|
|
return f"Error: {e}"
|
|
|
|
def stop_pod_container(self) -> str:
|
|
try:
|
|
container = self._get_podman_client().containers.get(self.safe_name)
|
|
container.stop()
|
|
self.status = "offline"
|
|
self.last_log = f"Stopped container {container.id}"
|
|
self.sync_status()
|
|
self.save()
|
|
return f"Container stopped successfully: {container.id}"
|
|
except podman.errors.NotFound:
|
|
self.last_log = f"Container '{self.safe_name}' not found"
|
|
self.save()
|
|
return f"Error: Container '{self.safe_name}' not found"
|
|
except Exception as e:
|
|
self.last_log = f"Error stopping container: {str(e)}"
|
|
self.save()
|
|
return f"Error: {e}"
|
|
|
|
def remove_pod_container(self) -> str:
|
|
try:
|
|
container = self._get_podman_client().containers.get(self.safe_name)
|
|
container.remove(force=True)
|
|
self.status = "offline"
|
|
self.container_id = None
|
|
self.last_log = f"Removed container {self.safe_name}"
|
|
self.sync_status()
|
|
self.save()
|
|
return f"Container removed successfully: {self.safe_name}"
|
|
except podman.errors.NotFound:
|
|
self.last_log = f"Container '{self.safe_name}' not found"
|
|
self.save()
|
|
return f"Error: Container '{self.safe_name}' not found"
|
|
except Exception as e:
|
|
self.last_log = f"Error removing container: {str(e)}"
|
|
self.save()
|
|
return f"Error: {e}" |