Module: Web API Details¶

Tasks¶

Background task management for long-running operations

`CancelToken` ¶

Source code in webapi/api/tasks.py

class CancelToken:
    def __init__(self):
        """ Thread-safe cancellation token. This basically just has a lock and a status
            indicating whether the job should be cancelled or not. Jobs use this to figure
            out whether the web cancel button has been clicked.
        """
        self._cancelled = False
        self._lock = threading.Lock()

    def cancel(self):
        with self._lock:
            self._cancelled = True

    def is_cancelled(self):
        with self._lock:
            return self._cancelled

`init()` ¶

Thread-safe cancellation token. This basically just has a lock and a status indicating whether the job should be cancelled or not. Jobs use this to figure out whether the web cancel button has been clicked.

Source code in webapi/api/tasks.py

def __init__(self):
    """ Thread-safe cancellation token. This basically just has a lock and a status
        indicating whether the job should be cancelled or not. Jobs use this to figure
        out whether the web cancel button has been clicked.
    """
    self._cancelled = False
    self._lock = threading.Lock()

`cancel_job(job_id, timeout=10.0)` ¶

Cancel a running job

Parameters:

Name	Type	Description	Default
`job_id`	`str`	ID of the job to cancel	required
`timeout`	`float`	Maximum time to wait for cancellation (seconds)	`10.0`

Returns:

Type	Description
`Dict[str, Any]`	Dict with success status and message

Source code in webapi/api/tasks.py

def cancel_job(job_id: str, timeout: float = 10.0) -> Dict[str, Any]:
    """
    Cancel a running job

    Args:
        job_id: ID of the job to cancel
        timeout: Maximum time to wait for cancellation (seconds)

    Returns:
        Dict with success status and message
    """
    job = jobs.get(job_id)
    if not job:
        return {'success': False, 'error': 'Job not found'}

    if job['status'] != 'running':
        return {'success': False, 'error': f"Job is not running (status: {job['status']})"}

    # Signal cancellation
    cancel_token = job.get('cancel_token')
    if cancel_token:
        cancel_token.cancel()

    # Wait for thread to finish
    thread = job.get('thread')
    if thread and thread.is_alive():
        thread.join(timeout=timeout)

        if thread.is_alive():
            # Thread didn't stop in time
            return {
                'success': False,
                'error': f'Job did not stop within {timeout} seconds',
                'partial_results': job.get('result')
            }

    # Get partial results from progress
    partial_results = {}
    if job.get('progress_obj'):
        partial_results = job['progress_obj'].get_snapshot()

    return {
        'success': True,
        'message': 'Job cancelled successfully',
        'partial_results': partial_results
    }

`get_job_status(job_id)` ¶

Get the status of a job

Source code in webapi/api/tasks.py

def get_job_status(job_id: str) -> Dict[str, Any]:
    """Get the status of a job"""
    job = jobs.get(job_id)
    if not job:
        return {'status': 'not_found'}

    result = {
        'job_id': job_id,
        'status': job['status'],
        'started_at': job.get('started_at')
    }

    # Add progress if running or complete (for final status)
    if job.get('progress_obj'):
        result['progress'] = job['progress_obj'].get_snapshot()

    # Add result if complete or cancelled
    if job['status'] in ['complete', 'cancelled']:
        result['result'] = job.get('result')

    # Add error if failed
    if job['status'] == 'error':
        result['error'] = job.get('error')

    return result

`run_async(func, *args, progress_obj=FetchProgress(), **kwargs)` ¶

Run a function asynchronously and return a job ID

Parameters:

Name	Type	Description	Default
`func`	`Callable`	Function to execute.	required
`*args`	`list`	Positional arguments for the function	`()`
`progress_obj`	`FetchProgress`	Optional progress tracking object	`FetchProgress()`
`**kwargs`	`list`	Keyword arguments for the function	`{}`

Returns:

Type	Description
`str`	Job ID string

Source code in webapi/api/tasks.py

def run_async(func: Callable, *args,
    progress_obj:FetchProgress=FetchProgress(),
    **kwargs) -> str:
    """
    Run a function asynchronously and return a job ID

    Args:
        func: Function to execute.
        *args (list): Positional arguments for the function
        progress_obj: Optional progress tracking object
        **kwargs (list): Keyword arguments for the function

    Returns:
        Job ID string
    """
    job_id = str(uuid4())
    cancel_token = CancelToken()

    jobs[job_id] = {
        'status': 'running',
        'progress_obj': progress_obj,
        'cancel_token': cancel_token,
        'thread': None,
        'result': None,
        'error': None,
        'started_at': datetime.now().isoformat()
    }

    def worker():
        """ The code that a
        """
        try:
            sig = inspect.signature(func)
            params = sig.parameters

            # Build kwargs based on what function accepts
            call_kwargs = dict(kwargs)
            if 'cancel_token' in params:
                call_kwargs['cancel_token'] = cancel_token
            if 'progress' in params and progress_obj is not None:
                call_kwargs['progress'] = progress_obj

            result = func(*args, **call_kwargs)

            # Check if cancelled during execution
            if cancel_token.is_cancelled():
                jobs[job_id]['status'] = 'cancelled'
                jobs[job_id]['result'] = result  # Partial results
            else:
                jobs[job_id]['status'] = 'complete'
                jobs[job_id]['result'] = result

            jobs[job_id]['error'] = None
            jobs[job_id]['started_at'] = jobs[job_id]['started_at']
        except Exception as e:
            jobs[job_id] = {
                'status': 'error',
                'progress_obj': progress_obj,
                'cancel_token': cancel_token,
                'thread': None,
                'result': None,
                'error': str(e),
                'started_at': jobs[job_id]['started_at']
            }

    thread = threading.Thread(target=worker, daemon=True)
    jobs[job_id]['thread'] = thread
    thread.start()

    return job_id

Progress¶

Progress tracking for long-running fetch operations

`FetchProgress` ¶

Thread-safe progress tracker for fetch operations

Source code in webapi/api/progress.py

class FetchProgress:
    """Thread-safe progress tracker for fetch operations"""

    def __init__(self, max_toots: int = 2000):
        self.servers_todo = 0
        self.servers_done = 0
        self.servers_fail = 0
        self.total_toots = 0
        self.max_toots = max_toots
        self.current_server: Optional[str] = None
        self.server_status: Dict[str, dict] = {}  # uri -> {'state': str, 'toots': int}
        self._lock = threading.Lock()

    def update(self, todo: int, done: int, fail: int, toots: int, current: Optional[str] = None):
        """Update progress metrics (thread-safe)"""
        with self._lock:
            self.servers_todo = todo
            self.servers_done = done
            self.servers_fail = fail
            self.total_toots = toots
            self.current_server = current

    def start_server(self, uri: str):
        """Mark a server as started (thread-safe)"""
        with self._lock:
            self.server_status[uri] = {'state': 'running', 'toots': 0}

    def set_pending(self, uri: str):
        """Mark a server as pending (thread-safe)"""
        with self._lock:
            self.server_status[uri] = {'state': 'pending', 'toots': 0}

    def update_server_progress(self, uri: str, toot_count: int):
        """Update toot count for a running server (thread-safe)"""
        with self._lock:
            if uri in self.server_status:
                self.server_status[uri]['toots'] = toot_count

    def complete_server(self, uri: str, final_count: int, failed: bool = False):
        """Mark a server as completed or failed (thread-safe)"""
        with self._lock:
            self.server_status[uri] = {
                'state': 'failed' if failed else 'done',
                'toots': final_count if not failed else 0
            }

    def get_snapshot(self) -> dict:
        """Get current progress as a dict (thread-safe)"""
        with self._lock:
            total = self.servers_done + self.servers_fail + self.servers_todo
            completed = self.servers_done + self.servers_fail
            progress_percent = int((completed / total * 100)) if total > 0 else 0

            return {
                'servers_todo': self.servers_todo,
                'servers_done': self.servers_done,
                'servers_fail': self.servers_fail,
                'total_servers': total,
                'completed_servers': completed,
                'progress_percent': progress_percent,
                'total_toots': self.total_toots,
                'max_toots': self.max_toots,
                'current_server': self.current_server,
                'server_status': dict(self.server_status)  # Copy for safety
            }

`complete_server(uri, final_count, failed=False)` ¶

Mark a server as completed or failed (thread-safe)

Source code in webapi/api/progress.py

def complete_server(self, uri: str, final_count: int, failed: bool = False):
    """Mark a server as completed or failed (thread-safe)"""
    with self._lock:
        self.server_status[uri] = {
            'state': 'failed' if failed else 'done',
            'toots': final_count if not failed else 0
        }

`get_snapshot()` ¶

Get current progress as a dict (thread-safe)

Source code in webapi/api/progress.py

def get_snapshot(self) -> dict:
    """Get current progress as a dict (thread-safe)"""
    with self._lock:
        total = self.servers_done + self.servers_fail + self.servers_todo
        completed = self.servers_done + self.servers_fail
        progress_percent = int((completed / total * 100)) if total > 0 else 0

        return {
            'servers_todo': self.servers_todo,
            'servers_done': self.servers_done,
            'servers_fail': self.servers_fail,
            'total_servers': total,
            'completed_servers': completed,
            'progress_percent': progress_percent,
            'total_toots': self.total_toots,
            'max_toots': self.max_toots,
            'current_server': self.current_server,
            'server_status': dict(self.server_status)  # Copy for safety
        }

`set_pending(uri)` ¶

Mark a server as pending (thread-safe)

Source code in webapi/api/progress.py

def set_pending(self, uri: str):
    """Mark a server as pending (thread-safe)"""
    with self._lock:
        self.server_status[uri] = {'state': 'pending', 'toots': 0}

`start_server(uri)` ¶

Mark a server as started (thread-safe)

Source code in webapi/api/progress.py

def start_server(self, uri: str):
    """Mark a server as started (thread-safe)"""
    with self._lock:
        self.server_status[uri] = {'state': 'running', 'toots': 0}

`update(todo, done, fail, toots, current=None)` ¶

Update progress metrics (thread-safe)

Source code in webapi/api/progress.py

def update(self, todo: int, done: int, fail: int, toots: int, current: Optional[str] = None):
    """Update progress metrics (thread-safe)"""
    with self._lock:
        self.servers_todo = todo
        self.servers_done = done
        self.servers_fail = fail
        self.total_toots = toots
        self.current_server = current

`update_server_progress(uri, toot_count)` ¶

Update toot count for a running server (thread-safe)

Source code in webapi/api/progress.py

def update_server_progress(self, uri: str, toot_count: int):
    """Update toot count for a running server (thread-safe)"""
    with self._lock:
        if uri in self.server_status:
            self.server_status[uri]['toots'] = toot_count

Routes¶

API routes for Mastoscore Web API