SQ-821: Announcements to users via CLI and webpage

docker/divbase_compose.yaml

@@ -162,7 +162,7 @@ services:
       - ../packages/divbase-api/src/divbase_api/migrations:/app/migrations # this is to get the migration scripts which are generated inside the container 
       - ../packages/divbase-api/src/divbase_api/alembic.ini:/app/alembic.ini
     healthcheck:
-      test: ["CMD", "curl", "-f", "http://localhost:8000/api/health"]
+      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/core/health"]
       interval: 5s
       timeout: 5s
       retries: 5

docs/development/database-migrations.md

@@ -65,6 +65,12 @@ op.execute("DROP TYPE IF EXISTS projectroles")
 
 See the bottom of the first migration script (2025-12-04_initial_migration.py) for an example of this.
 
+You could also run the automatic migration tests now to help verify your migrations scripts are correct.
+
+```bash
+pytest -s tests/migrations
+```
+
 #### 4. Test out the migration
 
 - In local development, migrations are applied automatically when the stack starts up using the `db-migrator` init-container.

packages/divbase-api/src/divbase_api/admin_panel.py

@@ -22,6 +22,7 @@
     BooleanField,
     DateTimeField,
     EmailField,
+    EnumField,
     HasOne,
     IntegerField,
     JSONField,
@@ -36,12 +37,13 @@
 from divbase_api.db import get_db
 from divbase_api.deps import _authenticate_frontend_user_from_tokens
 from divbase_api.frontend_routes.auth import get_login, post_logout
+from divbase_api.models.announcements import AnnouncementDB, AnnouncementLevel, AnnouncementTarget
 from divbase_api.models.project_versions import ProjectVersionDB
-from divbase_api.models.projects import ProjectDB, ProjectMembershipDB
-from divbase_api.models.revoked_tokens import RevokedTokenDB
+from divbase_api.models.projects import ProjectDB, ProjectMembershipDB, ProjectRoles
+from divbase_api.models.revoked_tokens import RevokedTokenDB, TokenRevokeReason
 from divbase_api.models.task_history import CeleryTaskMeta, TaskHistoryDB, TaskStartedAtDB
 from divbase_api.models.users import UserDB
-from divbase_api.security import get_password_hash
+from divbase_api.security import TokenType, get_password_hash
 
 logger = logging.getLogger(__name__)
 
@@ -275,7 +277,7 @@ class ProjectMembershipView(ModelView):
         IntegerField("id", label="ID", disabled=True),
         HasOne("user", identity="user", label="User"),
         HasOne("project", identity="project", label="Project"),
-        StringField("role", label="Role", required=True),
+        EnumField("role", label="Role", required=True, enum=ProjectRoles),
         DateTimeField("created_at", label="Created At", disabled=True),
         DateTimeField("updated_at", label="Updated At", disabled=True),
     ]
@@ -377,9 +379,9 @@ class RevokedTokenView(ModelView):
     fields = [
         IntegerField("id", label="ID", disabled=True),
         StringField("token_jti", label="Token JTI", required=True),
-        StringField("token_type", label="Token Type", required=True),
+        EnumField("token_type", label="Token Type", required=True, enum=TokenType),
         DateTimeField("revoked_at", label="Revoked At", disabled=True),
-        StringField("revoked_reason", label="Revoked Reason", required=True),
+        EnumField("revoked_reason", label="Revoke Reason", required=True, enum=TokenRevokeReason),
         IntegerField("user_id", label="User ID", required=False),
         HasOne("user", identity="user", label="User"),
         DateTimeField("created_at", label="Created At", disabled=True),
@@ -567,6 +569,37 @@ def can_delete(self, request: Request) -> bool:
         return False
 
 
+class AnnouncementView(ModelView):
+    """
+    Custom admin panel View for the AnnouncementDB model.
+    """
+
+    page_size_options = PAGINATION_DEFAULTS
+    fields = [
+        IntegerField("id", label="ID", disabled=True),
+        StringField("heading", label="Heading", required=True),
+        TextAreaField("message", label="Message", required=False),
+        EnumField("target", label="Target", required=True, enum=AnnouncementTarget),
+        EnumField("level", label="Level", required=True, enum=AnnouncementLevel),
+        DateTimeField(
+            "auto_expire_at",
+            help_text="Optional timestamp when the announcement will auto expire. After this timestamp the announcement will no longer be displayed on the frontend or CLI. Value can be left empty.",
+        ),
+        DateTimeField(
+            "created_at", help_text="Timestamp when the entry was created. Value determined by system.", disabled=True
+        ),
+        DateTimeField(
+            "updated_at",
+            help_text="Timestamp when the entry was last updated. Value determined by system.",
+            disabled=True,
+        ),
+    ]
+
+    exclude_fields_from_list = ["message"]
+    exclude_fields_from_edit = ["id", "created_at", "updated_at"]
+    exclude_fields_from_detail = []
+
+
 class DivBaseAuthProvider(AuthProvider):
     """
     This class enables starlette-admin to make use of DivBase's pre-existing auth system.
@@ -646,6 +679,8 @@ def register_admin_panel(app: FastAPI, engine: AsyncEngine) -> None:
     admin.add_view(
         CeleryTaskMetaView(CeleryTaskMeta, icon="fas fa-tasks", label="Celery Task Meta", identity="celery-meta")
     )
-
     admin.add_view(TaskStartedAtView(TaskStartedAtDB, icon="fas fa-clock", label="Task Started At"))
+    admin.add_view(
+        AnnouncementView(AnnouncementDB, icon="fas fa-bullhorn", label="Announcements", identity="announcement")
+    )
     admin.mount_to(app)

packages/divbase-api/src/divbase_api/crud/announcements.py

@@ -0,0 +1,31 @@
+"""
+Crud operations on the announcements table,
+which stores announcements that can be displayed to users on the frontend and the cli.
+
+Starlette admin will manage the creation/editing/deletion of announcements,
+so this module only covers retrieving active announcements to be displayed on frontend or by CLI.
+"""
+
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from divbase_api.models.announcements import AnnouncementDB, AnnouncementTarget
+from divbase_lib.api_schemas.announcements import AnnouncementResponse
+
+
+async def get_active_announcements(db: AsyncSession, target: AnnouncementTarget) -> list[AnnouncementResponse]:
+    """Get active announcements for a given target (cli, web or both)."""
+
+    if target == AnnouncementTarget.BOTH:
+        raise ValueError("Target cannot be both when retrieving announcements. Please specify either cli or web.")
+
+    stmt = (
+        select(AnnouncementDB)
+        .where((AnnouncementDB.target == target) | (AnnouncementDB.target == AnnouncementTarget.BOTH))
+        .where((AnnouncementDB.auto_expire_at.is_(None)) | (AnnouncementDB.auto_expire_at > datetime.now(timezone.utc)))
+    )
+    result = await db.execute(stmt)
+    announcements = result.scalars().all()
+    return [AnnouncementResponse.model_validate(announcement) for announcement in announcements]

packages/divbase-api/src/divbase_api/divbase_api.py

@@ -27,6 +27,7 @@
 from divbase_api.frontend_routes.projects import fr_projects_router
 from divbase_api.routes.admin import admin_router
 from divbase_api.routes.auth import auth_router
+from divbase_api.routes.core import core_router
 from divbase_api.routes.project_versions import project_version_router
 from divbase_api.routes.queries import query_router
 from divbase_api.routes.s3 import s3_router
@@ -66,23 +67,20 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
 app = FastAPI(lifespan=lifespan, title="DivBase API", docs_url="/api/v1/docs")
 
-
 app.include_router(auth_router, prefix="/api/v1/auth", tags=["auth"])
 app.include_router(admin_router, prefix="/api/v1/admin", tags=["admin"])
-app.include_router(s3_router, prefix="/api/v1/s3", tags=["s3"])
+app.include_router(core_router, prefix="/api/v1/core", tags=["core"])
 app.include_router(project_version_router, prefix="/api/v1/project-versions", tags=["project-versioning"])
-
+app.include_router(query_router, prefix="/api/v1/query", tags=["query"])
+app.include_router(s3_router, prefix="/api/v1/s3", tags=["s3"])
+app.include_router(task_history_router, prefix="/api/v1/task-history", tags=["task-history"])
+app.include_router(vcf_dimensions_router, prefix="/api/v1/vcf-dimensions", tags=["vcf-dimensions"])
 
 app.include_router(fr_auth_router, prefix="", include_in_schema=False)
 app.include_router(fr_core_router, prefix="", include_in_schema=False)
 app.include_router(fr_profile_router, prefix="/profile", include_in_schema=False)
 app.include_router(fr_projects_router, prefix="/projects", include_in_schema=False)
 
-app.include_router(query_router, prefix="/api/v1/query", tags=["query"])
-
-app.include_router(vcf_dimensions_router, prefix="/api/v1/vcf-dimensions", tags=["vcf-dimensions"])
-
-app.include_router(task_history_router, prefix="/api/v1/task-history", tags=["task-history"])
 
 register_exception_handlers(app)
 register_admin_panel(app=app, engine=engine)
@@ -91,12 +89,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 app.mount("/static", StaticFiles(directory=static_dir_path), name="static")
 
 
-# TODO - move below routes into routes dir when ready.
-@app.get("/api/health")
-def health():
-    return {"status": "ok"}
-
-
 def main():
     uvicorn.run("divbase_api.divbase_api:app", host="127.0.0.1", port=8000, reload=True)
 

packages/divbase-api/src/divbase_api/frontend_routes/core.py

@@ -10,9 +10,13 @@
 from fastapi import APIRouter, Depends, Request
 from fastapi.responses import HTMLResponse
 from fastapi.templating import Jinja2Templates
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from divbase_api.api_config import settings
+from divbase_api.crud.announcements import get_active_announcements
+from divbase_api.db import get_db
 from divbase_api.deps import get_current_user_from_cookie_optional
+from divbase_api.models.announcements import AnnouncementTarget
 from divbase_api.models.users import UserDB
 
 fr_core_router = APIRouter()
@@ -23,10 +27,15 @@
 
 
 @fr_core_router.get("/", response_class=HTMLResponse)
-async def get_home_page(request: Request, current_user: UserDB | None = Depends(get_current_user_from_cookie_optional)):
+async def get_home_page(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    current_user: UserDB | None = Depends(get_current_user_from_cookie_optional),
+):
     """Render the home page."""
+    announcements = await get_active_announcements(db=db, target=AnnouncementTarget.WEB)
     return templates.TemplateResponse(
         request=request,
         name="index.html",
-        context={"request": request, "current_user": current_user},
+        context={"request": request, "current_user": current_user, "announcements": announcements},
     )

packages/divbase-api/src/divbase_api/migrations/versions/2026-02-09_add_annoucements_db.py

@@ -0,0 +1,56 @@
+"""add annoucements db
+
+Revision ID: 2ec979fc9dbb
+Revises: 3e168ddc857e
+Create Date: 2026-02-09 09:43:52.822149
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "2ec979fc9dbb"
+down_revision: Union[str, Sequence[str], None] = "3e168ddc857e"
+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.create_table(
+        "announcement",
+        sa.Column("heading", sa.String(length=200), nullable=False),
+        sa.Column("message", sa.String(length=1000), nullable=True),
+        sa.Column("target", sa.Enum("BOTH", "CLI", "WEB", name="announcementtarget"), nullable=False),
+        sa.Column("level", sa.Enum("INFO", "SUCCESS", "WARNING", "DANGER", name="announcementlevel"), nullable=False),
+        sa.Column("auto_expire_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_announcement_heading"), "announcement", ["heading"], unique=False)
+    op.create_index(op.f("ix_announcement_id"), "announcement", ["id"], unique=False)
+    op.create_index(op.f("ix_announcement_level"), "announcement", ["level"], unique=False)
+    op.create_index(op.f("ix_announcement_target"), "announcement", ["target"], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f("ix_announcement_target"), table_name="announcement")
+    op.drop_index(op.f("ix_announcement_level"), table_name="announcement")
+    op.drop_index(op.f("ix_announcement_id"), table_name="announcement")
+    op.drop_index(op.f("ix_announcement_heading"), table_name="announcement")
+    op.drop_table("announcement")
+    # ### end Alembic commands ###
+
+    # Human added, explicitly drop enums
+    # see https://github.com/sqlalchemy/alembic/issues/886
+    op.execute("DROP TYPE IF EXISTS announcementtarget")
+    op.execute("DROP TYPE IF EXISTS announcementlevel")

packages/divbase-api/src/divbase_api/models/__init__.py

@@ -4,6 +4,7 @@
 They are imported in dependency order to avoid circular import issues.
 """
 
+from divbase_api.models.announcements import AnnouncementDB, AnnouncementTarget
 from divbase_api.models.base import Base, BaseDBModel
 from divbase_api.models.project_versions import ProjectVersionDB
 from divbase_api.models.projects import ProjectDB, ProjectMembershipDB, ProjectRoles
@@ -26,4 +27,6 @@
     "TokenRevokeReason",
     "RevokedTokenDB",
     "ProjectVersionDB",
+    "AnnouncementDB",
+    "AnnouncementTarget",
 ]

packages/divbase-api/src/divbase_api/models/announcements.py

@@ -0,0 +1,52 @@
+"""
+Announcements DB Model.
+
+Stores announcements that can be displayed to users on the frontend (banner) or via the CLI.
+"""
+
+from datetime import datetime
+from enum import StrEnum
+
+from sqlalchemy import DateTime, Enum, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from divbase_api.models.base import BaseDBModel
+
+
+class AnnouncementTarget(StrEnum):
+    """Possible targets for announcements."""
+
+    BOTH = "both"
+    CLI = "cli"
+    WEB = "web"
+
+
+class AnnouncementLevel(StrEnum):
+    """
+    Possible levels for announcements.
+    These match bootstrap alert levels, so this will control the announcement styling on the frontend.
+    """
+
+    INFO = "info"
+    SUCCESS = "success"
+    WARNING = "warning"
+    DANGER = "danger"
+
+
+class AnnouncementDB(BaseDBModel):
+    """
+    DB Model for an announcement.
+
+    id, created_at and updated_at are inherited from BaseDBModel.
+    """
+
+    __tablename__ = "announcement"
+
+    heading: Mapped[str] = mapped_column(String(200), index=True)
+    message: Mapped[str | None] = mapped_column(String(1000), nullable=True)
+    target: Mapped[AnnouncementTarget] = mapped_column(Enum(AnnouncementTarget), index=True)
+    level: Mapped[AnnouncementLevel] = mapped_column(Enum(AnnouncementLevel), index=True)
+    auto_expire_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    def __repr__(self) -> str:
+        return f"<AnnouncementDB id={self.id}, heading={self.heading}, target={self.target}, level={self.level}>"

packages/divbase-api/src/divbase_api/routes/core.py

@@ -0,0 +1,31 @@
+"""
+Core API routes for divbase, including health checks and announcements.
+
+Note that unlike every other API route these routes are not behind authentication...
+"""
+
+import logging
+
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from divbase_api.crud.announcements import get_active_announcements
+from divbase_api.db import get_db
+from divbase_api.models.announcements import AnnouncementTarget
+from divbase_lib.api_schemas.announcements import AnnouncementResponse
+
+logger = logging.getLogger(__name__)
+
+core_router = APIRouter()
+
+
+@core_router.get("/health", status_code=status.HTTP_200_OK, response_model=dict[str, str])
+def health():
+    """Basic health check endpoint for the server."""
+    return {"status": "ok"}
+
+
+@core_router.get("/announcements", status_code=status.HTTP_200_OK, response_model=list[AnnouncementResponse])
+async def announcements(db: AsyncSession = Depends(get_db)):
+    """Returns active announcements for CLI users from the server."""
+    return await get_active_announcements(db=db, target=AnnouncementTarget.CLI)