96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
import socket
|
|
from typing import Optional, List, Dict
|
|
from app.config.database import db
|
|
|
|
|
|
class PortManager:
|
|
"""Service for managing port allocation for script interfaces."""
|
|
|
|
def __init__(self, port_range_start: int = 5200, port_range_end: int = 5400):
|
|
self.port_range_start = port_range_start
|
|
self.port_range_end = port_range_end
|
|
self.allocated_ports = set()
|
|
self._load_allocated_ports()
|
|
|
|
def _load_allocated_ports(self):
|
|
"""Load currently allocated ports from database."""
|
|
# This would query the port_allocations table if implemented
|
|
# For now, we'll use in-memory tracking
|
|
pass
|
|
|
|
def is_port_available(self, port: int) -> bool:
|
|
"""Check if a specific port is available."""
|
|
# Check if port is in our allocated range (5200-5400)
|
|
if port < 5200 or port > 5400:
|
|
return False
|
|
|
|
if port in self.allocated_ports:
|
|
return False
|
|
|
|
# Check if port is actually free on the system
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.settimeout(1)
|
|
s.bind(("127.0.0.1", port))
|
|
return True
|
|
except socket.error:
|
|
return False
|
|
|
|
def allocate_port(
|
|
self, script_id: int, user_id: int, project_id: int
|
|
) -> Optional[int]:
|
|
"""Allocate an available port for a script instance."""
|
|
for port in range(self.port_range_start, self.port_range_end + 1):
|
|
if self.is_port_available(port):
|
|
self.allocated_ports.add(port)
|
|
|
|
# TODO: Store allocation in database
|
|
# port_allocation = PortAllocation(
|
|
# port_number=port,
|
|
# script_id=script_id,
|
|
# user_id=user_id,
|
|
# project_id=project_id,
|
|
# status='allocated'
|
|
# )
|
|
# db.session.add(port_allocation)
|
|
# db.session.commit()
|
|
|
|
return port
|
|
|
|
return None
|
|
|
|
def release_port(self, port: int):
|
|
"""Release a port back to the available pool."""
|
|
if port in self.allocated_ports:
|
|
self.allocated_ports.remove(port)
|
|
|
|
# TODO: Update database
|
|
# port_allocation = PortAllocation.query.filter_by(
|
|
# port_number=port, status='allocated'
|
|
# ).first()
|
|
# if port_allocation:
|
|
# port_allocation.status = 'released'
|
|
# port_allocation.released_at = db.func.now()
|
|
# db.session.commit()
|
|
|
|
def get_allocated_ports(self) -> List[int]:
|
|
"""Get list of currently allocated ports."""
|
|
return list(self.allocated_ports)
|
|
|
|
def get_available_ports_count(self) -> int:
|
|
"""Get count of available ports."""
|
|
total_ports = self.port_range_end - self.port_range_start + 1
|
|
return total_ports - len(self.allocated_ports)
|
|
|
|
def cleanup_orphaned_ports(self):
|
|
"""Clean up ports that are no longer in use."""
|
|
orphaned_ports = []
|
|
|
|
for port in self.allocated_ports.copy():
|
|
if self.is_port_available(port):
|
|
# Port is not actually in use, release it
|
|
orphaned_ports.append(port)
|
|
self.allocated_ports.remove(port)
|
|
|
|
return orphaned_ports
|