Guardrail: BanList API Management

backend/app/alembic/versions/004_added_ban_list_config.py

@@ -0,0 +1,39 @@
+"""Added ban_list table
+
+Revision ID: 004
+Revises: 003
+Create Date: 2026-02-05 09:42:54.128852
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+from sqlalchemy.dialects import postgresql
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '004'
+down_revision: Union[str, Sequence[str], None] = "003"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table('ban_list',
+        sa.Column('id', sa.Uuid(), nullable=False),
+        sa.Column('name', sa.String(), nullable=False),
+        sa.Column('description', sa.String(), nullable=False),
+        sa.Column('org_id', sa.Integer(), nullable=False),
+        sa.Column('project_id', sa.Integer(), nullable=False),
+        sa.Column('domain', sa.String(), nullable=False),
+        sa.Column('is_public', sa.Boolean(), nullable=False, server_default=sa.false()),
+        sa.Column("banned_words", postgresql.ARRAY(sa.String()), nullable=False, server_default="{}"),
+        sa.Column('created_at', sa.DateTime(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), nullable=False),
+
+        sa.PrimaryKeyConstraint('id'),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table('validator_config')

backend/app/api/main.py

@@ -1,11 +1,12 @@
 from fastapi import APIRouter
 
-from app.api.routes import utils, guardrails, validator_configs
+from app.api.routes import ban_list_configs, guardrails, utils, validator_configs
 
 api_router = APIRouter()
 api_router.include_router(utils.router)
 api_router.include_router(guardrails.router)
 api_router.include_router(validator_configs.router)
+api_router.include_router(ban_list_configs.router)
 
 # if settings.ENVIRONMENT == "local":
 #     api_router.include_router(private.router)

backend/app/api/routes/ban_list_configs.py

@@ -0,0 +1,118 @@
+from typing import List, Optional
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlmodel import Session
+
+from app.api.deps import AuthDep, SessionDep
+from app.crud.ban_list_crud import ban_list_crud
+from app.schemas.ban_list_config import (
+    BanListCreate,
+    BanListUpdate,
+    BanListResponse
+)
+
+router = APIRouter(
+    prefix="/guardrails/ban-lists", 
+    tags=["Ban Lists"]
+)
+
+
+def check_owner(obj, org_id, project_id):
+    if obj.org_id != org_id or obj.project_id != project_id:
+        raise HTTPException(status_code=403, detail="Not owner")
+
+
+@router.post(
+        "/", 
+        response_model=BanListResponse
+    )
+def create_ban_list(
+    payload: BanListCreate,
+    session: SessionDep,
+    org_id: int,
+    project_id: int,
+    _: AuthDep,
+):
+    return ban_list_crud.create(
+        session,
+        data=payload,
+        org_id=org_id,
+        project_id=project_id,
+    )
+
+
+@router.get(
+        "/", 
+        response_model=list[BanListResponse]
+    )
+def list_ban_lists(
+    org_id: int,
+    project_id: int,
+    session: SessionDep,
+    _: AuthDep,
+    domain: Optional[str] = None,
+):
+    return ban_list_crud.list(
+        session,
+        org_id=org_id,
+        project_id=project_id,
+        domain=domain,
+    )
+
+
+@router.get(
+        "/{id}",
+        response_model=BanListResponse
+    )
+def get_ban_list(
+    id: UUID,
+    org_id: int,
+    project_id: int,
+    session: SessionDep,
+    _: AuthDep,
+):
+    obj = ban_list_crud.get(session, id)
+    if not obj:
+        raise HTTPException(404)
+
+    if not obj.is_public:
+        check_owner(obj, org_id, project_id)
+    return obj
+
+
+@router.patch(
+        "/{id}",
+        response_model=BanListResponse
+    )
+def update_ban_list(
+    id: UUID,
+    org_id: int,
+    project_id: int,
+    payload: BanListUpdate,
+    session: SessionDep,
+    _: AuthDep,
+):
+    obj = ban_list_crud.get(session, id)
+    if not obj:
+        raise HTTPException(404)
+
+    check_owner(obj, org_id, project_id)
+    return ban_list_crud.update(session, obj=obj, data=payload)
+
+
+@router.delete("/{id}")
+def delete_ban_list(
+    id: UUID,
+    org_id: int,
+    project_id: int,
+    session: SessionDep,
+    _: AuthDep,
+):
+    obj = ban_list_crud.get(session, id)
+    if not obj:
+        raise HTTPException(404)
+
+    check_owner(obj, org_id, project_id)
+    ban_list_crud.delete(session, obj)
+    return {"success": True}

backend/app/crud/ban_list_crud.py

@@ -0,0 +1,79 @@
+from datetime import datetime
+from typing import List, Optional
+from uuid import UUID
+
+from sqlmodel import Session, select
+
+from app.models.config.ban_list_table import BanList
+from app.schemas.ban_list_config import BanListCreate, BanListUpdate
+from app.utils import now
+
+class BanListCrud:
+
+    def create(     
+        self,
+        db: Session,
+        *,
+        data: BanListCreate,
+        org_id: int,
+        project_id: int,
+    ) -> BanList:
+        obj = BanList(
+            **data.model_dump(),
+            org_id=org_id,
+            project_id=project_id,
+        )
+        db.add(obj)
+        db.commit()
+        db.refresh(obj)
+        return obj
+
+    def get(self, db: Session, id: UUID) -> Optional[BanList]:
+        return db.get(BanList, id)
+
+    def list(
+        self,
+        db: Session,
+        *,
+        org_id: int,
+        project_id: int,
+        domain: Optional[str] = None,
+    ) -> List[BanList]:
+        stmt = select(BanList).where(
+            (
+                (BanList.org_id == org_id) &
+                (BanList.project_id == project_id)
+            ) |
+            (BanList.is_public == True)
+        )
+
+        if domain:
+            stmt = stmt.where(BanList.domain == domain)
+
+        return list(db.exec(stmt))
+
+    def update(
+        self,
+        db: Session,
+        *,
+        obj: BanList,
+        data: BanListUpdate,
+    ) -> BanList:
+        update_data = data.model_dump(exclude_unset=True)
+
+        for k, v in update_data.items():
+            setattr(obj, k, v)
+
+        obj.updated_at = now()
+
+        db.add(obj)
+        db.commit()
+        db.refresh(obj)
+        return obj
+
+    def delete(self, db: Session, obj: BanList):
+        db.delete(obj)
+        db.commit()
+
+
+ban_list_crud = BanListCrud()

backend/app/models/config/ban_list_table.py

@@ -0,0 +1,75 @@
+from datetime import datetime
+from typing import List, Optional
+from uuid import UUID, uuid4
+
+from sqlalchemy import Column, String
+from sqlalchemy.dialects.postgresql import ARRAY
+from sqlmodel import Field, SQLModel
+
+from app.utils import now
+
+class BanList(SQLModel, table=True):
+    __tablename__ = "ban_list"
+
+    id: UUID = Field(
+        default_factory=uuid4, 
+        primary_key=True, 
+        index=True,
+        sa_column_kwargs={"comment": "Unique identifier for the ban list entry"}
+    )
+
+    name: str = Field(
+        nullable=False, 
+        sa_column_kwargs={"comment": "Name of the ban list entry"}
+    )
+
+    description: Optional[str] = Field(
+        nullable=False,
+        sa_column_kwargs={"comment": "Description of the ban list entry"}
+    )
+
+    banned_words: list[str] = Field(
+        default_factory=list,
+        sa_column=Column(
+            ARRAY(String),
+            nullable=False,
+            comment="List of banned words",
+        ), 
+        description=("List of banned words")
+    )
+
+    org_id: int = Field(
+        index=True,
+        nullable=False,
+        sa_column_kwargs={"comment": "Identifier for the organization"},
+    )
+
+    project_id: int = Field(
+        index=True,
+        nullable=False,
+        sa_column_kwargs={"comment": "Identifier for the project"},
+    )
+
+    domain: str = Field(
+        default=None, 
+        index=False,
+        nullable=False,
+        sa_column_kwargs={"comment": "Domain or context for the ban list entry"}
+    )
+
+    is_public: bool = Field(
+        default=False,
+        sa_column_kwargs={"comment": "Whether the ban list entry is public or private"}
+    )
+
+    created_at: datetime = Field(
+        default_factory=now,
+        nullable=False,
+        sa_column_kwargs={"comment": "Timestamp when the ban list entry was created"}
+    )
+
+    updated_at: datetime = Field(
+        default_factory=now,
+        nullable=False,
+        sa_column_kwargs={"comment": "Timestamp when the ban list entry was last updated"}
+    )

backend/app/schemas/ban_list_config.py

@@ -0,0 +1,33 @@
+from uuid import UUID
+from datetime import datetime
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+from sqlmodel import SQLModel
+
+class BanListBase(SQLModel):
+    name: str
+    description: str
+    banned_words: list[str]
+    domain: str
+    is_public: bool = False
+
+
+class BanListCreate(BanListBase):
+    pass
+
+
+class BanListUpdate(SQLModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+    banned_words: Optional[list[str]] = None
+    domain: Optional[str] = None
+    is_public: Optional[bool] = None
+
+
+class BanListResponse(BanListBase):
+    id: UUID
+    org_id: int
+    project_id: int
+    created_at: datetime
+    updated_at: datetime