gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into copilot/add-document-similarity-detection

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-03-01 22:06:55 +01:00
committed by GitHub
parent b7c78177e9 698c6a222e
commit 1e1ba1d8d7
17 changed files with 1096 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/__init__.py
+2
View File
@@ -24,6 +24,7 @@ from app.api.url_upload import router as url_upload_router
# Import all the individual routers
from app.api.user import router as user_router
from app.api.webhooks import router as webhooks_router
# Set up logging
logger = logging.getLogger(__name__)
@@ -48,3 +49,4 @@ router.include_router(search_router)
router.include_router(queue_router)
router.include_router(saved_searches_router)
router.include_router(similarity_router)
router.include_router(webhooks_router)
app/api/webhooks.py
+206
View File
@@ -0,0 +1,206 @@
"""API endpoints for managing webhook configurations.
Provides CRUD operations for webhook configs that notify external systems
when document events occur (``document.uploaded``, ``document.processed``,
``document.failed``).
"""
import json
import logging
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import WebhookConfig
from app.utils.webhook import VALID_EVENTS
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/webhooks", tags=["webhooks"])
DbSession = Annotated[Session, Depends(get_db)]
# ---------------------------------------------------------------------------
# Auth helper (reuse the pattern from settings API)
# ---------------------------------------------------------------------------
def _require_admin(request: Request) -> dict:
"""Ensure the caller is an admin. Raises 403 otherwise."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
# ---------------------------------------------------------------------------
# Pydantic schemas
# ---------------------------------------------------------------------------
class WebhookCreate(BaseModel):
"""Schema for creating a new webhook configuration."""
url: str = Field(..., min_length=1, max_length=2048, description="Target URL for webhook delivery")
secret: str | None = Field(default=None, max_length=512, description="Shared secret for HMAC-SHA256 signatures")
events: list[str] = Field(..., min_length=1, description="List of events to subscribe to")
is_active: bool = Field(default=True, description="Whether the webhook is active")
description: str | None = Field(default=None, max_length=500, description="Optional human-readable description")
class WebhookUpdate(BaseModel):
"""Schema for updating an existing webhook configuration."""
url: str | None = Field(default=None, min_length=1, max_length=2048)
secret: str | None = Field(default=None, max_length=512)
events: list[str] | None = Field(default=None, min_length=1)
is_active: bool | None = None
description: str | None = Field(default=None, max_length=500)
class WebhookResponse(BaseModel):
"""Schema returned to clients (secret is never exposed)."""
id: int
url: str
events: list[str]
is_active: bool
description: str | None
has_secret: bool
model_config = {"from_attributes": True}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _validate_events(events: list[str]) -> None:
"""Raise 422 if any event name is not recognised."""
invalid = set(events) - VALID_EVENTS
if invalid:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"Invalid event(s): {', '.join(sorted(invalid))}. Valid events: {', '.join(sorted(VALID_EVENTS))}",
)
def _to_response(cfg: WebhookConfig) -> dict[str, Any]:
"""Convert a DB model instance to a response dict."""
try:
events = json.loads(cfg.events)
except (json.JSONDecodeError, TypeError):
events = []
return {
"id": cfg.id,
"url": cfg.url,
"events": events,
"is_active": cfg.is_active,
"description": cfg.description,
"has_secret": cfg.secret is not None and len(cfg.secret) > 0,
}
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("/", summary="List all webhook configurations")
def list_webhooks(db: DbSession, _admin: AdminUser) -> list[dict[str, Any]]:
"""Return all webhook configurations. Secrets are never included."""
configs = db.query(WebhookConfig).order_by(WebhookConfig.id).all()
return [_to_response(c) for c in configs]
@router.get("/{webhook_id}", summary="Get a single webhook configuration")
def get_webhook(webhook_id: int, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Return a single webhook configuration by ID."""
cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
if not cfg:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
return _to_response(cfg)
@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a webhook configuration")
def create_webhook(body: WebhookCreate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Create a new webhook configuration."""
_validate_events(body.events)
cfg = WebhookConfig(
url=body.url,
secret=body.secret,
events=json.dumps(sorted(body.events)),
is_active=body.is_active,
description=body.description,
)
try:
db.add(cfg)
db.commit()
db.refresh(cfg)
except Exception:
db.rollback()
raise
logger.info("Webhook %d created for events %s", cfg.id, body.events)
return _to_response(cfg)
@router.put("/{webhook_id}", summary="Update a webhook configuration")
def update_webhook(webhook_id: int, body: WebhookUpdate, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Update an existing webhook configuration. Only supplied fields are changed."""
cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
if not cfg:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
if body.url is not None:
cfg.url = body.url
if body.secret is not None:
cfg.secret = body.secret
if body.events is not None:
_validate_events(body.events)
cfg.events = json.dumps(sorted(body.events))
if body.is_active is not None:
cfg.is_active = body.is_active
if body.description is not None:
cfg.description = body.description
try:
db.commit()
db.refresh(cfg)
except Exception:
db.rollback()
raise
logger.info("Webhook %d updated", cfg.id)
return _to_response(cfg)
@router.delete("/{webhook_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a webhook configuration")
def delete_webhook(webhook_id: int, db: DbSession, _admin: AdminUser) -> None:
"""Delete a webhook configuration."""
cfg = db.query(WebhookConfig).filter(WebhookConfig.id == webhook_id).first()
if not cfg:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Webhook not found")
try:
db.delete(cfg)
db.commit()
except Exception:
db.rollback()
raise
logger.info("Webhook %d deleted", cfg.id)
@router.get("/events/", summary="List valid webhook event types")
def list_events(_admin: AdminUser) -> list[str]:
"""Return the list of valid event types that can be subscribed to."""
return sorted(VALID_EVENTS)
app/config.py
+6
View File
@@ -274,6 +274,12 @@ class Settings(BaseSettings):
description="Send notifications when files are successfully processed",
)
# Webhook settings
webhook_enabled: bool = Field(
default=True,
description="Enable webhook delivery for document events",
)
# File upload size limits (for security - see SECURITY_AUDIT.md)
max_upload_size: int = Field(
default=1073741824, # 1GB in bytes (1024 * 1024 * 1024)
app/models.py
+15
View File
@@ -146,3 +146,18 @@ class SavedSearch(Base):
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
__table_args__ = (UniqueConstraint("user_id", "name", name="unique_user_search_name"),)
class WebhookConfig(Base):
"""Webhook configuration for notifying external systems of document events."""
__tablename__ = "webhook_configs"
id = Column(Integer, primary_key=True, index=True)
url = Column(String, nullable=False) # Target URL for webhook delivery
secret = Column(String, nullable=True) # Shared secret for HMAC-SHA256 signature
events = Column(Text, nullable=False) # JSON list of subscribed events
is_active = Column(Boolean, default=True, nullable=False) # Whether the webhook is active
description = Column(String, nullable=True) # Optional human-readable description
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
app/tasks/webhook_tasks.py
+39
View File
@@ -0,0 +1,39 @@
"""Celery task for asynchronous webhook delivery with retry and backoff.
Uses :class:`~app.tasks.retry_config.BaseTaskWithRetry` so failed deliveries
are automatically retried with exponential backoff (default: 60 s, 300 s,
900 s) and ±20 % jitter.
"""
import logging
from typing import Any
from app.celery_app import celery
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils.webhook import deliver_webhook
logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry, bind=True, name="webhook.deliver")
def deliver_webhook_task(self, url: str, payload: dict[str, Any], secret: str | None = None) -> dict[str, Any]:
"""Deliver a webhook payload to *url* with automatic retries.
Args:
url: Target webhook URL.
payload: The full webhook payload envelope.
secret: Optional shared secret for HMAC-SHA256 signing.
Returns:
A dict with ``status`` and ``url`` on success.
Raises:
RuntimeError: Re-raised to trigger Celery retry on delivery failure.
"""
logger.info("Delivering webhook to %s (attempt %d/%d)", url, self.request.retries + 1, self.max_retries + 1)
success = deliver_webhook(url, payload, secret)
if success:
return {"status": "delivered", "url": url}
raise RuntimeError(f"Webhook delivery to {url} failed")
app/utils/webhook.py
+165
View File
@@ -0,0 +1,165 @@
"""Webhook delivery utility for notifying external systems of document events.
Provides functions to dispatch webhook payloads with HMAC-SHA256 signatures
and to query active webhook configurations from the database.
Supported events:
- ``document.uploaded`` – a new document has been ingested
- ``document.processed`` – a document finished processing successfully
- ``document.failed`` – document processing failed
"""
import hashlib
import hmac
import json
import logging
import time
from typing import Any
import requests
from app.database import SessionLocal
from app.models import WebhookConfig
logger = logging.getLogger(__name__)
#: Events recognised by the webhook subsystem.
VALID_EVENTS: frozenset[str] = frozenset(
{
"document.uploaded",
"document.processed",
"document.failed",
}
)
#: Timeout (seconds) for outgoing webhook HTTP requests.
WEBHOOK_TIMEOUT = 10
def compute_signature(payload_bytes: bytes, secret: str) -> str:
"""Compute an HMAC-SHA256 hex-digest for *payload_bytes* using *secret*.
Args:
payload_bytes: The raw JSON body to sign.
secret: The shared secret string.
Returns:
``sha256=<hex-digest>`` signature string.
"""
mac = hmac.new(secret.encode("utf-8"), payload_bytes, hashlib.sha256)
return f"sha256={mac.hexdigest()}"
def deliver_webhook(url: str, payload: dict[str, Any], secret: str | None = None) -> bool:
"""Send a single webhook POST request.
Args:
url: Target URL.
payload: JSON-serialisable dictionary.
secret: If provided, an ``X-Webhook-Signature`` header is included.
Returns:
``True`` when the remote server responds with a 2xx status.
"""
body = json.dumps(payload, default=str, sort_keys=True)
body_bytes = body.encode("utf-8")
headers: dict[str, str] = {"Content-Type": "application/json"}
if secret:
headers["X-Webhook-Signature"] = compute_signature(body_bytes, secret)
try:
resp = requests.post(url, data=body_bytes, headers=headers, timeout=WEBHOOK_TIMEOUT)
if resp.ok:
logger.info("Webhook delivered to %s (status %d)", url, resp.status_code)
return True
logger.warning("Webhook to %s returned status %d", url, resp.status_code)
return False
except requests.RequestException as exc:
logger.error("Webhook delivery to %s failed: %s", url, exc)
return False
def build_payload(event: str, data: dict[str, Any]) -> dict[str, Any]:
"""Build a standardised webhook payload envelope.
Args:
event: The event name (e.g. ``document.uploaded``).
data: Event-specific data.
Returns:
Dictionary with ``event``, ``timestamp``, and ``data`` keys.
"""
return {
"event": event,
"timestamp": time.time(),
"data": data,
}
def get_active_webhooks_for_event(event: str) -> list[dict[str, Any]]:
"""Return all active webhook configs subscribed to *event*.
Queries the database directly so this helper can be called from both the
API layer and Celery tasks.
Args:
event: The event name to filter on.
Returns:
A list of dicts with ``id``, ``url``, ``secret``, and ``events`` keys.
"""
db = SessionLocal()
try:
configs = db.query(WebhookConfig).filter(WebhookConfig.is_active.is_(True)).all()
result: list[dict[str, Any]] = []
for cfg in configs:
try:
subscribed = json.loads(cfg.events)
except (json.JSONDecodeError, TypeError):
subscribed = []
if event in subscribed:
result.append(
{
"id": cfg.id,
"url": cfg.url,
"secret": cfg.secret,
"events": subscribed,
}
)
return result
finally:
db.close()
def dispatch_webhook_event(event: str, data: dict[str, Any]) -> None:
"""Fan-out a webhook event to all matching active configurations.
This is the main entry-point used by application code to trigger webhooks.
It delegates to :func:`deliver_webhook_task` (Celery) for each matching
webhook so delivery happens asynchronously with automatic retries.
Args:
event: Event name (must be in :data:`VALID_EVENTS`).
data: Event-specific payload data.
"""
if event not in VALID_EVENTS:
logger.warning("Ignoring unknown webhook event: %s", event)
return
webhooks = get_active_webhooks_for_event(event)
if not webhooks:
logger.debug("No active webhooks for event %s", event)
return
payload = build_payload(event, data)
# Import here to avoid circular dependency with celery_app
from app.tasks.webhook_tasks import deliver_webhook_task
for wh in webhooks:
try:
deliver_webhook_task.delay(wh["url"], payload, wh["secret"])
logger.debug("Queued webhook delivery to %s for event %s", wh["url"], event)
except Exception as exc:
logger.error("Failed to queue webhook to %s: %s", wh["url"], exc)