Write a queue service of blueapi calls

[stan-dot]

certainly use this

https://github.com/DiamondLightSource/mx-daq-ui/blob/main/src/blueapi/blueapi.ts

[stan-dot]

backed by redis? might be joined with scan metadata service

[stan-dot]

initial impl

from fastapi import FastAPI, Depends, WebSocket, WebSocketDisconnect, HTTPException, Query
from fastapi.security import OAuth2AuthorizationCodeBearer
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
import redis
import json
import asyncio
import datetime

# Redis connection
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

# OAuth2 OIDC Authentication
OIDC_ISSUER = "https://your-oidc-provider.com"
OIDC_CLIENT_ID = "your-client-id"
OIDC_CLIENT_SECRET = "your-client-secret"

oauth2_scheme = OAuth2AuthorizationCodeBearer(
    tokenUrl=f"{OIDC_ISSUER}/token",
    authorizationUrl=f"{OIDC_ISSUER}/auth"
)

app = FastAPI()

# Store active WebSocket connections
connections = {}

class QueueItem(BaseModel):
    id: str
    data: Dict[str, Any]
    played_at: Optional[str] = None

def get_queue_key(user_id: str) -> str:
    return f"queue:{user_id}"

def get_past_queue_key(user_id: str) -> str:
    return f"past_queue:{user_id}"

def notify_clients(user_id: str):
    """Notify WebSocket clients about queue changes."""
    if user_id in connections:
        queue_data = redis_client.lrange(get_queue_key(user_id), 0, -1)
        asyncio.create_task(connections[user_id].send_text(json.dumps(queue_data)))

@app.get("/queue", response_model=List[QueueItem])
def get_queue(token: str = Depends(oauth2_scheme)):
    """Get the user's queue."""
    user_id = token  # Assuming the user ID is in the token
    queue_data = redis_client.lrange(get_queue_key(user_id), 0, -1)
    return [json.loads(item) for item in queue_data]

@app.post("/queue")
def add_to_queue(item: QueueItem, token: str = Depends(oauth2_scheme)):
    """Add an item to the user's queue."""
    user_id = token
    redis_client.rpush(get_queue_key(user_id), json.dumps(item.dict()))
    notify_clients(user_id)
    return {"message": "Item added"}

@app.delete("/queue/{item_id}")
def remove_from_queue(item_id: str, token: str = Depends(oauth2_scheme)):
    """Remove an item from the user's queue by ID."""
    user_id = token
    queue_key = get_queue_key(user_id)
    queue_data = redis_client.lrange(queue_key, 0, -1)
    
    updated_queue = [item for item in queue_data if json.loads(item)['id'] != item_id]
    
    redis_client.delete(queue_key)
    for item in updated_queue:
        redis_client.rpush(queue_key, item)
    
    notify_clients(user_id)
    return {"message": "Item removed"}

@app.post("/queue/play/{item_id}")
def play_item(item_id: str, token: str = Depends(oauth2_scheme)):
    """Mark an item as played, moving it to the past queue with a timestamp."""
    user_id = token
    queue_key = get_queue_key(user_id)
    past_queue_key = get_past_queue_key(user_id)
    queue_data = redis_client.lrange(queue_key, 0, -1)
    
    played_item = None
    updated_queue = []
    for item in queue_data:
        item_dict = json.loads(item)
        if item_dict['id'] == item_id:
            played_item = item_dict
        else:
            updated_queue.append(item)
    
    if played_item:
        played_item["played_at"] = datetime.datetime.utcnow().isoformat()
        redis_client.rpush(past_queue_key, json.dumps(played_item))
        redis_client.delete(queue_key)
        for item in updated_queue:
            redis_client.rpush(queue_key, item)
        notify_clients(user_id)
        return {"message": "Item played and moved to past queue"}
    
    raise HTTPException(status_code=404, detail="Item not found")

@app.get("/queue/past", response_model=List[QueueItem])
def get_past_queue(
    token: str = Depends(oauth2_scheme),
    page: int = Query(1, alias="page", ge=1),
    page_size: int = Query(10, alias="page_size", ge=1, le=100),
    from_timestamp: Optional[str] = Query(None, alias="from_timestamp"),
    to_timestamp: Optional[str] = Query(None, alias="to_timestamp")
):
    """Get the user's past queue with pagination and timestamp filtering."""
    user_id = token
    past_queue_data = redis_client.lrange(get_past_queue_key(user_id), 0, -1)
    parsed_data = [json.loads(item) for item in past_queue_data]
    
    if from_timestamp or to_timestamp:
        filtered_data = []
        for item in parsed_data:
            played_at = datetime.datetime.fromisoformat(item["played_at"])
            if from_timestamp:
                from_dt = datetime.datetime.fromisoformat(from_timestamp)
                if played_at < from_dt:
                    continue
            if to_timestamp:
                to_dt = datetime.datetime.fromisoformat(to_timestamp)
                if played_at > to_dt:
                    continue
            filtered_data.append(item)
    else:
        filtered_data = parsed_data
    
    start_idx = (page - 1) * page_size
    end_idx = start_idx + page_size
    paginated_data = filtered_data[start_idx:end_idx]
    
    return paginated_data

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str):
    """WebSocket endpoint for real-time queue updates."""
    await websocket.accept()
    user_id = token
    connections[user_id] = websocket
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        del connections[user_id]

[stan-dot]

waiting for feedback on metadata service

[stan-dot]

let's just postpone it as much as possible. from issue #57 looks like this might not be as necessary. raw requests should be fine, wrapped in a gui like mx-daq-ui have