Fw/implement job status endpoint

alembic/versions/53a4dee6ef34_add_passed_and_feedback_column_to_.py

@@ -0,0 +1,33 @@
+"""add passed and feedback column to GradingJob + add title to TestCase
+
+Revision ID: 53a4dee6ef34
+Revises: 9a72ad7167bf
+Create Date: 2025-10-09 14:36:41.006723
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '53a4dee6ef34'
+down_revision: Union[str, Sequence[str], None] = '9a72ad7167bf'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('test_case', sa.Column('title', sa.TEXT(), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('test_case', 'title')
+    # ### end Alembic commands ###

app/api/schema/grading.py

@@ -15,3 +15,14 @@ class GradingResult(BaseModel):
     penalty: datetime
     feedback: str
     hint: Optional[str]
+
+
+class GradingJobRead(BaseModel):
+    id: str
+    tan_code: str
+    exercise_id: int
+    status: str
+    started: datetime
+    terminated: datetime | None
+    passed: bool | None
+    feedback: GradingResult | None

app/api/v1/grading.py

@@ -3,15 +3,15 @@
 import logging
 from uuid import uuid4
 
-import amqp
-from amqp import Channel
+from aio_pika import Message
+from aio_pika.abc import AbstractRobustChannel
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi import status
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.api.schema.grading import ExerciseSubmission
-from app.config import GRADING_RESPONSE_QUEUE_TTL
+from app.api.schema.grading import ExerciseSubmission, GradingJobRead
+from app.config import MESSAGE_QUEUE_EXCHANGE_NAME
 from app.db.database import get_session
 from app.db.model.exercise import ExerciseProgress
 from app.db.model.grading import GradingJob
@@ -23,11 +23,7 @@
 )
 
 
-async def submit_grading_job(job_msg: dict, session: AsyncSession, ch: Channel):
-    ch.queue_declare(queue=f'grading_response.{job_msg["job_id"]}', durable=True, arguments={
-        "x-expires": GRADING_RESPONSE_QUEUE_TTL,
-    })
-
+async def submit_grading_job(job_msg: dict, session: AsyncSession, ch: AbstractRobustChannel):
     session.add(GradingJob(
         id=job_msg["job_id"],
         tan_code=job_msg["tan_code"],
@@ -37,12 +33,14 @@ async def submit_grading_job(job_msg: dict, session: AsyncSession, ch: Channel):
     ))
 
     job_msg["job_id"] = str(job_msg["job_id"])
-    ch.basic_publish(amqp.Message(json.dumps(job_msg)), routing_key='grading_jobs')
+
+    exchange = await ch.get_exchange(MESSAGE_QUEUE_EXCHANGE_NAME)
+    await exchange.publish(Message(body=json.dumps(job_msg).encode('utf-8')), routing_key='grading_jobs')
 
 
 @router.post("/", status_code=status.HTTP_201_CREATED, response_model=str)
 async def create_submission(new_submission: ExerciseSubmission, session: AsyncSession = Depends(get_session),
-                            mq_channel: Channel = Depends(get_mq_channel)) -> str:
+                            mq_channel: AbstractRobustChannel = Depends(get_mq_channel)) -> str:
     stmt = select(ExerciseProgress).where(ExerciseProgress.exercise_id == new_submission.exercise_id,
                                           ExerciseProgress.tan_code == new_submission.tan_code,
                                           ExerciseProgress.end_time.is_(None))
@@ -69,3 +67,12 @@ async def create_submission(new_submission: ExerciseSubmission, session: AsyncSe
         logging.error(e)
         await session.rollback()
         raise HTTPException(status_code=500, detail=f"Scheduling a grading job failed. {str(e)}")
+
+
+@router.get("/{job_id}",
+            response_model=GradingJobRead,
+            status_code=status.HTTP_200_OK)
+async def get_submission_status(job_id: str, session: AsyncSession = Depends(get_session)) -> GradingJobRead:
+    stmt = select(GradingJob).where(GradingJob.id == job_id)
+    result = await session.execute(stmt)
+    return GradingJobRead(**result.scalars().first().to_dict())

app/config.py

@@ -7,6 +7,6 @@
 DATABASE_URL = os.getenv("BLOCKSEMBLER_DB_URI", "postgresql+asyncpg://postgres:postgres@localhost:5432/blocksembler")
 
 MESSAGE_QUEUE_URL = os.environ.get('BLOCKSEMBLER_MESSAGE_QUEUE_URL', 'localhost')
-MESSAGE_QUEUE_USER = os.environ.get('BLOCKSEMBLER_MESSAGE_QUEUE_USER', 'blocksembler')
-MESSAGE_QUEUE_PASSWORD = os.environ.get('BLOCKSEMBLER_MESSAGE_QUEUE_PASSWORD', 'blocksembler')
-GRADING_RESPONSE_QUEUE_TTL = os.environ.get('BLOCKSEMBLER_GRADING_RESPONSE_QUEUE_TTL', 1000 * 60 * 15)
+MESSAGE_QUEUE_USER = os.getenv("BLOCKSEMBLER_MESSAGE_QUEUE_USER", "blocksembler")
+MESSAGE_QUEUE_PASSWORD = os.getenv("BLOCKSEMBLER_MESSAGE_QUEUE_PASSWORD", "blocksembler")
+MESSAGE_QUEUE_EXCHANGE_NAME = os.getenv("BLOCKSEMBLER_MESSAGE_QUEUE_EXCHANGE_NAME", "blocksembler-grading-exchange")

app/db/model/grading.py

@@ -14,3 +14,15 @@ class GradingJob(Base):
     terminated = sa.Column(sa.DateTime(timezone=True), nullable=True)
     passed = sa.Column(sa.BOOLEAN, nullable=True)
     feedback = sa.Column(sa.JSON, nullable=True)
+
+    def to_dict(self):
+        return {
+            "id": str(self.id),
+            "tan_code": self.tan_code,
+            "exercise_id": self.exercise_id,
+            "status": self.status,
+            "started": self.started,
+            "terminated": self.terminated,
+            "passed": self.passed,
+            "feedback": self.feedback
+        }

app/main.py

@@ -1,7 +1,6 @@
 import logging
 from contextlib import asynccontextmanager
 
-import amqp
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 

app/mq/message_queue.py

@@ -1,8 +1,10 @@
-from amqp import Connection
+import aio_pika
+from aio_pika.abc import AbstractRobustChannel
 
-from app.config import MESSAGE_QUEUE_URL, MESSAGE_QUEUE_USER, MESSAGE_QUEUE_PASSWORD
 
+async def get_mq_channel() -> AbstractRobustChannel:
+    connection = await aio_pika.connect_robust("amqp://blocksembler:blocksembler@localhost:5672")
 
-async def get_mq_channel() -> Connection:
-    with Connection(MESSAGE_QUEUE_URL, userid=MESSAGE_QUEUE_USER, password=MESSAGE_QUEUE_PASSWORD) as c:
-        yield c.channel()  # noqa
+    async with connection:
+        channel = await connection.channel()
+        yield channel

requirements.txt

@@ -4,5 +4,5 @@ sqlalchemy~=2.0.0
 alembic~=1.16.5
 asyncpg~=0.30.0
 aiosqlite~=0.21.0
-amqp~=5.3.1
-pytest~=8.4.2
+pytest~=8.4.2
+aio-pika~=9.5.7

tests/test_submission.py

@@ -1,18 +1,24 @@
 import asyncio
-from unittest.mock import MagicMock, ANY
+from unittest.mock import MagicMock, ANY, AsyncMock
 
-from amqp import Connection
+from aio_pika.abc import AbstractRobustChannel
 from fastapi import status
 from fastapi.testclient import TestClient
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
 
-from app.config import GRADING_RESPONSE_QUEUE_TTL
 from app.db.database import get_session
 from app.main import app
 from app.mq.message_queue import get_mq_channel
 from tests.util.db_util import create_test_tables, get_override_dependency, insert_all_records, DB_URI
 
 
+def setup_mocks():
+    channel_mock = MagicMock()
+    exchange_mock = MagicMock()
+    exchange_mock.publish = AsyncMock()
+    channel_mock.get_exchange = AsyncMock(return_value=exchange_mock)
+    return channel_mock, exchange_mock
+
 class TestSubmission:
     def setup_method(self):
         self.engine = create_async_engine(DB_URI, echo=True, future=True)
@@ -22,10 +28,10 @@ def setup_method(self):
         asyncio.run(insert_all_records(self.async_session))
 
     def test_post_submission(self):
-        mock_channel = MagicMock()
+        channel_mock, exchange_mock = setup_mocks()
 
-        async def get_mq_connection_override() -> Connection:
-            yield mock_channel  # noqa
+        async def get_mq_connection_override() -> AbstractRobustChannel:
+            yield channel_mock  # noqa
 
         app.dependency_overrides[get_session] = get_override_dependency(self.engine)
         app.dependency_overrides[get_mq_channel] = get_mq_connection_override
@@ -43,16 +49,13 @@ async def get_mq_connection_override() -> Connection:
         print(response.json())
 
         assert response.status_code == status.HTTP_201_CREATED
-        mock_channel.basic_publish.assert_called_once_with(ANY, routing_key='grading_jobs')
-        mock_channel.queue_declare.assert_called_once_with(queue=ANY, durable=True, arguments={
-            "x-expires": GRADING_RESPONSE_QUEUE_TTL,
-        })
+        exchange_mock.publish.assert_awaited_once_with(ANY, routing_key='grading_jobs')
 
     def test_post_invalid_submission(self):
-        mock_channel = MagicMock()
+        channel_mock, exchange_mock = setup_mocks()
 
-        async def get_mq_connection_override() -> Connection:
-            yield mock_channel  # noqa
+        async def get_mq_connection_override() -> AbstractRobustChannel:
+            yield channel_mock  # noqa
 
         app.dependency_overrides[get_session] = get_override_dependency(self.engine)
         app.dependency_overrides[get_mq_channel] = get_mq_connection_override
@@ -70,5 +73,5 @@ async def get_mq_connection_override() -> Connection:
         print(response.json())
 
         assert response.status_code == status.HTTP_400_BAD_REQUEST
-        mock_channel.basic_publish.assert_not_called()
-        mock_channel.queue_declare.assert_not_called()
+        channel_mock.get_exchange.assert_not_awaited()
+        exchange_mock.publish.assert_not_awaited()

tests/util/demo_data.py

@@ -45,7 +45,7 @@
     },
     {
         "id": 3,
-        "title": "Demo exercise 2",
+        "title": "Demo exercise 3",
         "markdown": "",
         "coding_mode": "bbp",
         "allow_skip_after": 0,