49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
from django.contrib.auth.models import User
|
|
from django_eventstream import send_event
|
|
from django_eventstream.channelmanager import DefaultChannelManager
|
|
|
|
from deployment.models import Deployment
|
|
|
|
|
|
def parse_channel(channel: str) -> tuple[str] | None:
|
|
parts = channel.lstrip("deployment_").split("_")
|
|
if len_part := len(parts):
|
|
if len_part == 1:
|
|
return (parts[0], "")
|
|
if len_part == 2:
|
|
return (parts[0], parts[1])
|
|
return
|
|
|
|
|
|
class DeploymentChannelManager(DefaultChannelManager):
|
|
def can_read_channel(self, user: User, channel: str):
|
|
match parse_channel(channel):
|
|
case (user_id, ""):
|
|
return user_id == user.id or user.is_superuser
|
|
case (user_id, _):
|
|
# TODO(rmanach): check if the deployment belongs to the user
|
|
return user_id == user.id or user.is_superuser
|
|
return False
|
|
|
|
|
|
class Event:
|
|
@staticmethod
|
|
def send_details(deployment: Deployment, progress: int):
|
|
send_event(
|
|
f"deployment_{deployment.user.id}_{deployment.id}",
|
|
"message",
|
|
{
|
|
"id": deployment.id,
|
|
"status": deployment.status,
|
|
"progress": progress,
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
def send(deployment: Deployment):
|
|
send_event(
|
|
f"deployment_{deployment.user.id}",
|
|
"message",
|
|
{"id": deployment.id, "status": deployment.status},
|
|
)
|