Refactor job queue from RQ-style to Pythonic state management

[Copilot]

The job queue implementation mimicked RQ's interface with multiple is_* properties (is_queued, is_started, is_finished, is_failed), each triggering separate database queries. This created unnecessary complexity and poor performance.

Changes

Replaced Job with BuildJob class

• Single get_state() method retrieves all state in one query instead of 4+ separate queries
• Explicit update_meta() and update_status() methods replace property setters
• Added state caching to eliminate redundant DB access
• Backward-compatible meta property and save_meta() for existing build.py code

Simplified API response building

• New build_job_response() function uses single get_state(refresh=True) call
• Removed return_job_v1() with its multiple property accesses
• Safe dictionary access to prevent KeyErrors on null metadata

Build function updates

• Serialize BuildRequest with model_dump() for JSON compatibility
• Updated test helper to support new API

Example

Before:

if job.is_failed:  # Query 1
    error = job.latest_result().exc_string  # Query 2
elif job.is_queued:  # Query 3
    position = job.get_position()  # Query 4, 5

After:

state = job.get_state(refresh=True)  # Single query
if state["status"] == "failed":
    error = state["exc_string"]
elif state["status"] == "queued":
    position = state["queue_position"]

Net: -10 lines, 1 query instead of 4+, clearer intent.

Original prompt

Objective

Refactor the job queue implementation in PR #5 to use a cleaner, more Pythonic design instead of mimicking RQ's interface. The current implementation works but has unnecessary complexity from trying to match RQ's API style.

Current Implementation Issues

1. Too many status properties: is_queued, is_started, is_finished, is_failed all require separate DB queries
2. Inefficient DB access: Each property getter opens a new session and queries the database
3. Complex meta handling: Using property setters/getters for meta dict is verbose
4. Over-engineered: Trying to match RQ's API when we don't need to

Proposed Refactoring

1. Simplify Job Class (asu/job_queue.py)

Replace the current Job class with a cleaner design:

class BuildJob:
    """Represents a firmware build job with simplified state management."""
    
    def __init__(self, job_id: str):
        self.id = job_id
        self._cached_state = None
    
    def get_state(self, refresh: bool = False) -> dict:
        """Get full job state in one query.
        
        Returns:
            dict with keys: id, status, meta, result, enqueued_at, started_at, 
            finished_at, exc_string, queue_position (if queued)
        """
        if self._cached_state is None or refresh:
            session = get_session()
            try:
                job_model = session.query(JobModel).filter_by(id=self.id).first()
                if not job_model:
                    return None
                
                state = {
                    "id": job_model.id,
                    "status": job_model.status,
                    "meta": job_model.meta or {},
                    "result": job_model.result,
                    "enqueued_at": job_model.enqueued_at,
                    "started_at": job_model.started_at,
                    "finished_at": job_model.finished_at,
                    "exc_string": job_model.exc_string,
                }
                
                # Calculate queue position if queued
                if job_model.status == "queued":
                    position = session.query(JobModel).filter(
                        JobModel.status == "queued",
                        JobModel.enqueued_at < job_model.enqueued_at
                    ).count()
                    state["queue_position"] = position
                else:
                    state["queue_position"] = None
                    
                self._cached_state = state
            finally:
                session.close()
        
        return self._cached_state
    
    def update_meta(self, meta_updates: dict):
        """Update job metadata efficiently.
        
        Args:
            meta_updates: Dictionary of metadata to update/add
        """
        session = get_session()
        try:
            job_model = session.query(JobModel).filter_by(id=self.id).first()
            if job_model:
                current_meta = job_model.meta or {}
                current_meta.update(meta_updates)
                job_model.meta = current_meta
                session.commit()
                # Invalidate cache
                self._cached_state = None
        finally:
            session.close()
    
    def update_status(self, status: str, **fields):
        """Update job status and optional fields.
        
        Args:
            status: New status (queued, started, finished, failed)
            **fields: Additional fields to update (result, exc_string, etc.)
        """
        session = get_session()
        try:
            job_model = session.query(JobModel).filter_by(id=self.id).first()
            if job_model:
                job_model.status = status
                for key, value in fields.items():
                    setattr(job_model, key, value)
                session.commit()
                # Invalidate cache
                self._cached_state = None
        finally:
            session.close()
    
    @property
    def status(self) -> str:
        """Get current job status."""
        state = self.get_state()
        return state["status"] if state else None
    
    @property
    def meta(self) -> dict:
        """Get job metadata."""
        state = self.get_state()
        return state["meta"] if state else {}

2. Simplify API Response Building (asu/routers/api.py)

Replace the return_job_v1() function with a cleaner implementation:

def build_job_response(job: BuildJob) -> tuple[dict, int, dict]:
    """Build API response from job state.
    
    Args:
        job: BuildJob instance
        
    Returns:
        Tuple of (response_dict, status_code, headers_dict)
    """
    state = job.get_state(refresh=True)
    
    if not state:
        return {"status": 404, "detail": "Job not found"}, 404, {}
    
    # Base response with metadata
    response = {
        "request_hash": state["id"],
        "enqueued_at": state["enqueued_at"],
        **state["...

</details>



<!-- START COPILOT CODING AGENT SUFFIX -->

*This pull request was created from Copilot chat.*
>

<!-- START COPILOT CODING AGENT TIPS -->
---

💡 You can make Copilot smarter by setting up custom instructions, customizing its development environment and configuring Model Context Protocol (MCP) servers. Learn more [Copilot coding agent tips](https://gh.io/copilot-coding-agent-tips) in the docs.