한 프로젝트의 실행 자리를 소유하는 합성 모듈

"""One-process project lanes with ownership and explicit stop-request state."""
from dataclasses import dataclass, field
from threading import RLock
import re

class LaneError(ValueError):pass

def _label(value):
    if not isinstance(value,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,79}',value):raise LaneError('INVALID_LABEL')

@dataclass(frozen=True,eq=False)
class Lease:
    project:str
    run:str
    _token:object=field(repr=False)

class ProjectLanes:
    def __init__(self):
        self._lock=RLock()
        self._active={}

    def claim(self,project,run):
        _label(project);_label(run)
        with self._lock:
            if project in self._active:raise LaneError('PROJECT_BUSY')
            lease=Lease(project,run,object())
            self._active[project]={'lease':lease,'stop_requested':False}
            return lease

    def _owned(self,lease):
        if type(lease) is not Lease:raise LaneError('INVALID_LEASE')
        item=self._active.get(lease.project)
        if not item or item['lease']._token is not lease._token or item['lease'].run!=lease.run:raise LaneError('LEASE_NOT_CURRENT')
        return item

    def request_stop(self,lease):
        with self._lock:
            item=self._owned(lease)
            item['stop_requested']=True
            return self.view(lease.project)

    def release(self,lease,outcome):
        if outcome not in ['succeeded','failed','cancelled','needs_review']:raise LaneError('INVALID_OUTCOME')
        with self._lock:
            item=self._owned(lease)
            receipt={'project':lease.project,'run':lease.run,'outcome':outcome,'stop_requested':item['stop_requested'],'work_executed_here':False}
            del self._active[lease.project]
            return receipt

    def view(self,project):
        _label(project)
        with self._lock:
            item=self._active.get(project)
            return {'project':project,'busy':bool(item),**({'run':item['lease'].run,'state':'stop_requested' if item['stop_requested'] else 'running'} if item else {})}