Fix streaming docker logs (#3)

* streaming docker logs
* enable pushing docker images from PRs
* removing build ghcr
* allowing only one cbio import at a time
* added more formatting & checks to ci
* added pyrefly typehecks

Author: stolpeo

.github/workflows/ci.yml

@@ -2,32 +2,47 @@ name: CI
 
 on:
   push:
-    branches: ["main"]
+    branches: ["**"]
   pull_request:
     branches: ["**"]
 
 jobs:
   lint:
     runs-on: ubuntu-latest
+
     container:
       image: ghcr.io/astral-sh/uv:python3.12-bookworm-slim
+
     steps:
       - uses: actions/checkout@v6
+
       - name: Install dependencies
         run: uv sync --all-extras
+
       - name: Lint
         run: uv run ruff check
 
+      - name: Format check
+        run: uv run ruff format --check
+
+      - name: Type check
+        run: uv run pyrefly check
+
   test:
     runs-on: ubuntu-latest
+
     container:
       image: ghcr.io/astral-sh/uv:python3.12-bookworm-slim
+
     steps:
       - uses: actions/checkout@v6
+
       - name: Install dependencies
         run: uv sync --all-extras
+
       - name: Test
         run: uv run pytest -v --cov=app --cov-report=xml --cov-report=term-missing --cov-fail-under=95
+
       - name: Upload coverage report
         uses: actions/upload-artifact@v4
         with:
@@ -37,21 +52,36 @@ jobs:
   docker:
     runs-on: ubuntu-latest
     needs: [lint, test]
+
     permissions:
       contents: read
       packages: write
+
     steps:
       - uses: actions/checkout@v6
+
       - name: Log in to GHCR
         uses: docker/login-action@v3
         with:
           registry: ghcr.io
           username: ${{ github.repository_owner }}
           password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Docker meta
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ghcr.io/${{ github.repository }}
+          tags: |
+            type=ref,event=branch
+            type=ref,event=pr
+            type=sha
+            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
+
       - name: Build and push
         uses: docker/build-push-action@v6
         with:
           context: .
           file: docker/Dockerfile
-          push: ${{ github.ref == 'refs/heads/main' }}
-          tags: ghcr.io/${{ github.repository }}:latest,ghcr.io/${{ github.repository }}:${{ github.sha }}
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}

Makefile

@@ -12,6 +12,11 @@ format:
 .PHONY: check
 check:
 	uv run ruff check .
+	uv run ruff format --check .
+
+.PHONY: type
+type:
+	uv run pyright
 
 .PHONY: fix
 fix:
@@ -41,10 +46,6 @@ worker:
 docker:
 	bash docker/build.sh
 
-.PHONY: docker-ghcr
-docker-ghcr:
-	bash docker/build-ghcr.sh
-
 .PHONY: db-migrate
 db-migrate:
 	uv run alembic upgrade head

app/__init__.py

@@ -1,8 +1,9 @@
 # cbio-ingest FastAPI application
-from dotenv import load_dotenv
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import metadata as _metadata
 
+from dotenv import load_dotenv
+
 load_dotenv()
 
 try:

app/fs.py

@@ -66,13 +66,9 @@ def get_fs_service_studies(
     session: Session = Depends(get_session),
 ) -> FileSystemService:
     """Dependency to get filesystem service instance."""
-    return FileSystemService(
-        session=session, base_path=os.getenv("STUDY_DIR", "/app/study")
-    )
+    return FileSystemService(session=session, base_path=os.getenv("STUDY_DIR", "/app/study"))
 
 
 def get_fs_service_panels(session: Session = Depends(get_session)) -> FileSystemService:
     """Dependency to get filesystem service instance."""
-    return FileSystemService(
-        session=session, base_path=os.getenv("PANEL_DIR", "/app/panel")
-    )
+    return FileSystemService(session=session, base_path=os.getenv("PANEL_DIR", "/app/panel"))

app/routers/panels.py

@@ -7,7 +7,7 @@
 from app.auth import verify_token
 from app.db import get_session
 from app.fs import FileSystemService, get_fs_service_panels
-from app.models import IngestQuery, Panel, Status
+from app.models import IngestQuery, Panel, Status, Study
 from app.scheduler import queue
 from app.tasks import ingest_panel
 
@@ -22,7 +22,11 @@ async def list_panels(
     fs: FileSystemService = Depends(get_fs_service_panels),
     token=Depends(verify_token),
 ) -> list[Panel]:
-    """List all ingested panels. Pass `?available` to list panels on disk instead. Pass `?all` to merge both."""
+    """List all ingested panels.
+
+    Pass `?available` to list panels on disk instead.
+    Pass `?all` to merge both.
+    """
     if available is not None:
         return fs.list_panels()
     if all is not None:
@@ -33,7 +37,11 @@ async def list_panels(
     return list(session.exec(select(Panel)).all())
 
 
-@router.post("/", status_code=201, responses={400: {"description": "Bad Request"}})
+@router.post(
+    "/",
+    status_code=201,
+    responses={400: {"description": "Bad Request"}, 409: {"description": "Conflict"}},
+)
 async def create_panel(
     data: IngestQuery,
     keep_logs: bool = Query(default=False),
@@ -65,6 +73,12 @@ async def create_panel(
         session.commit()
         session.refresh(panel)
 
+    if (
+        session.exec(select(Study).where(Study.status == Status.IN_PROGRESS)).first()
+        or session.exec(select(Panel).where(Panel.status == Status.IN_PROGRESS)).first()
+    ):
+        raise HTTPException(status_code=409, detail="Another ingestion is already in progress")
+
     job_timeout = int(os.getenv("JOB_TIMEOUT", "3600"))
     job = queue.enqueue(ingest_panel, panel.id, job_timeout=job_timeout)
 

app/routers/studies.py

@@ -1,13 +1,13 @@
 import os
 
-from sqlalchemy.orm import attributes
 from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import attributes
 from sqlmodel import Session, select
 
 from app.auth import verify_token
 from app.db import get_session
 from app.fs import FileSystemService, get_fs_service_studies
-from app.models import IngestQuery, Status, Study
+from app.models import IngestQuery, Panel, Status, Study
 from app.scheduler import queue
 from app.tasks import ingest_study
 
@@ -22,7 +22,11 @@ async def list_studies(
     fs: FileSystemService = Depends(get_fs_service_studies),
     token=Depends(verify_token),
 ) -> list[Study]:
-    """List all ingested studies. Pass `?available` to list studies on disk instead. Pass `?all` to merge both."""
+    """List all ingested studies.
+
+    Pass `?available` to list studies on disk instead.
+    Pass `?all` to merge both.
+    """
     if available is not None:
         return fs.list_studies()
     if all is not None:
@@ -33,7 +37,11 @@ async def list_studies(
     return list(session.exec(select(Study)).all())
 
 
-@router.post("/", status_code=201, responses={400: {"description": "Bad Request"}})
+@router.post(
+    "/",
+    status_code=201,
+    responses={400: {"description": "Bad Request"}, 409: {"description": "Conflict"}},
+)
 async def create_study(
     data: IngestQuery,
     keep_logs: bool = Query(default=False),
@@ -65,6 +73,12 @@ async def create_study(
         session.commit()
         session.refresh(study)
 
+    if (
+        session.exec(select(Study).where(Study.status == Status.IN_PROGRESS)).first()
+        or session.exec(select(Panel).where(Panel.status == Status.IN_PROGRESS)).first()
+    ):
+        raise HTTPException(status_code=409, detail="Another ingestion is already in progress")
+
     job_timeout = int(os.getenv("JOB_TIMEOUT", "3600"))
     job = queue.enqueue(ingest_study, study.id, job_timeout=job_timeout)
 

app/scheduler.py

@@ -7,6 +7,4 @@
 redis_port = int(os.getenv("REDIS_PORT", "6379"))
 redis_password = os.getenv("REDIS_PASSWORD") or None
 
-queue = Queue(
-    connection=Redis(host=redis_host, port=redis_port, password=redis_password)
-)
+queue = Queue(connection=Redis(host=redis_host, port=redis_port, password=redis_password))

app/tasks.py

@@ -1,3 +1,4 @@
+import codecs
 import os
 from datetime import UTC, datetime
 
@@ -48,42 +49,59 @@ def _run_ingest(
         add_log(entity, LogLevel.INFO, "worker", f"Running command: {' '.join(cmd)}")
         print(f"Starting ingestion with command: {' '.join(cmd)} ...")
 
-        exec_result = container.exec_run(
-            cmd=cmd,
-            workdir=workdir,
+        entity.command = " ".join(cmd)
+        session.add(entity)
+        session.commit()
+        session.refresh(entity)
+
+        exec_id = client.api.exec_create(
+            container.id,
+            cmd,
             stdout=True,
             stderr=True,
-            stream=False,
+            workdir=workdir,
+            environment={"PYTHONUNBUFFERED": "1"},
         )
+        stream = client.api.exec_start(exec_id["Id"], stream=True)
+        decoder = codecs.getincrementaldecoder("utf-8")()
+        buffer = ""
+
+        for chunk in stream:
+            text = decoder.decode(chunk)
+            buffer += text
+            while "\n" in buffer:
+                line, buffer = buffer.split("\n", 1)
+                line = line.rstrip("\r")
+
+                if not line.strip():
+                    continue
+
+                add_log(entity, LogLevel.INFO, "docker", line)
+                session.add(entity)
+                session.commit()
+
+        session.add(entity)
+        session.commit()
+        session.refresh(entity)
+
+        inspect = client.api.exec_inspect(exec_id["Id"])
+        exit_code = inspect["ExitCode"]
 
         container.reload()
-        entity.command = " ".join(cmd)
         entity.cbioportal_version = (
             getattr(container, "attrs", {}).get("Config", {}).get("Image", "unknown")
         )
         session.add(entity)
         session.commit()
         session.refresh(entity)
 
-        logs = exec_result.output.decode("utf-8").split("\n")
-        exit_code = exec_result.exit_code
-
         print(f"Finished ingestion with exit code {exit_code} (0 = success)")
 
-        log_level = LogLevel.INFO if exit_code == 0 else LogLevel.ERROR
-
-        for log in logs:
-            stripped_log = log.strip()
-            if stripped_log:
-                add_log(entity, log_level, "docker", stripped_log)
-
         if exit_code == 0:
             print("Restarting container to apply changes ...")
             container.restart()
             print("Finished restarting container")
-            add_log(
-                entity, LogLevel.INFO, "worker", "Container restarted to apply changes."
-            )
+            add_log(entity, LogLevel.INFO, "worker", "Container restarted to apply changes.")
             mark_completed(entity, session)
             entity.date_ingested = datetime.now(UTC)
             session.add(entity)
@@ -117,7 +135,7 @@ def ingest_study(study_id: int) -> None:
         cmd = [
             "metaImport.py",
             "-u",
-            os.getenv("CBIOPORTAL_URL", "http://cbioportal:8080"),
+            "http://cbioportal:8080",
             "-s",
             f"/study/{validated_name}",
             "-o",

app/tests/test_auth.py

@@ -12,17 +12,13 @@ class TestVerifyToken:
 
     def test_valid_token(self, mock_token: str):
         """Test that valid token is accepted."""
-        credentials = HTTPAuthorizationCredentials(
-            scheme="Bearer", credentials=mock_token
-        )
+        credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=mock_token)
         result = verify_token(credentials)
         assert result == mock_token
 
     def test_invalid_token(self, mock_token: str):
         """Test that invalid token raises HTTPException."""
-        credentials = HTTPAuthorizationCredentials(
-            scheme="Bearer", credentials="wrong-token"
-        )
+        credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials="wrong-token")
         with pytest.raises(HTTPException) as exc_info:
             verify_token(credentials)