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

Merge pull request #81 from christianlouis/copilot/slow-down-processall-endpoint

Browse Source 
Add configurable rate limiting to /processall endpoint
This commit is contained in:
Christian Krakau-Louis
2026-02-07 16:04:37 +01:00
committed by GitHub
parent c84f64a846 0de31fc775
commit 642bdb1e56
7 changed files with 397 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.demo
+5
View File
@@ -6,6 +6,11 @@ EXTERNAL_HOSTNAME=docuelevate.example.com
GOTENBERG_URL=http://gotenberg:3000
ALLOW_FILE_DELETE=true # Allow deletion of file records
# **Batch Processing Settings**
# Control throttling behavior for the /processall endpoint to prevent overwhelming downstream APIs
PROCESSALL_THROTTLE_THRESHOLD=20 # Number of files above which throttling is applied (default: 20)
PROCESSALL_THROTTLE_DELAY=3 # Delay in seconds between each task submission when throttling (default: 3)
# **Authentication**
AUTH_ENABLED=true
# Generate a secure random string, for example:
app/api/process.py
+68 -45
View File
@@ -1,26 +1,29 @@
"""
Document processing API endpoints
"""
from fastapi import APIRouter, HTTPException
import logging
import os
from fastapi import APIRouter, HTTPException
from app.api.common import resolve_file_path
from app.auth import require_login
from app.config import settings
from app.api.common import resolve_file_path
from app.tasks.process_document import process_document
from app.tasks.upload_to_dropbox import upload_to_dropbox
from app.tasks.upload_to_paperless import upload_to_paperless
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
from app.tasks.upload_to_google_drive import upload_to_google_drive
from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.send_to_all import send_to_all_destinations
from app.tasks.upload_to_dropbox import upload_to_dropbox
from app.tasks.upload_to_google_drive import upload_to_google_drive
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.upload_to_paperless import upload_to_paperless
# Set up logging
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/process/")
@require_login
def process(file_path: str):
@@ -28,96 +31,92 @@ def process(file_path: str):
file_path = resolve_file_path(file_path)
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = process_document.delay(file_path)
return {"task_id": task.id, "status": "queued"}
@router.post("/send_to_dropbox/")
@require_login
def send_to_dropbox_endpoint(file_path: str):
"""Send a document to Dropbox."""
file_path = resolve_file_path(file_path, 'processed')
file_path = resolve_file_path(file_path, "processed")
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = upload_to_dropbox.delay(file_path)
return {"task_id": task.id, "status": "queued"}
@router.post("/send_to_paperless/")
@require_login
def send_to_paperless_endpoint(file_path: str):
"""Send a document to Paperless-ngx."""
file_path = resolve_file_path(file_path, 'processed')
file_path = resolve_file_path(file_path, "processed")
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = upload_to_paperless.delay(file_path)
return {"task_id": task.id, "status": "queued"}
@router.post("/send_to_nextcloud/")
@require_login
def send_to_nextcloud_endpoint(file_path: str):
"""Send a document to NextCloud."""
file_path = resolve_file_path(file_path, 'processed')
file_path = resolve_file_path(file_path, "processed")
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = upload_to_nextcloud.delay(file_path)
return {"task_id": task.id, "status": "queued"}
@router.post("/send_to_google_drive/")
@require_login
def send_to_google_drive_endpoint(file_path: str):
"""Send a document to Google Drive."""
file_path = resolve_file_path(file_path, 'processed')
file_path = resolve_file_path(file_path, "processed")
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = upload_to_google_drive.delay(file_path)
return {"task_id": task.id, "status": "queued"}
@router.post("/send_to_onedrive/")
@require_login
def send_to_onedrive_endpoint(file_path: str):
"""Send a document to OneDrive."""
file_path = resolve_file_path(file_path, 'processed')
file_path = resolve_file_path(file_path, "processed")
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = upload_to_onedrive.delay(file_path)
return {"task_id": task.id, "status": "queued"}
@router.post("/send_to_all_destinations/")
@require_login
def send_to_all_destinations_endpoint(file_path: str):
"""Call the aggregator task that sends this file to all configured destinations."""
file_path = resolve_file_path(file_path, 'processed')
file_path = resolve_file_path(file_path, "processed")
if not os.path.exists(file_path):
raise HTTPException(
status_code=400, detail=f"File {file_path} not found."
)
raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
task = send_to_all_destinations.delay(file_path)
return {"task_id": task.id, "status": "queued", "file_path": file_path}
@router.post("/processall")
@require_login
def process_all_pdfs_in_workdir():
"""Finds all .pdf files in <workdir> and enqueues them for processing."""
"""
Finds all .pdf files in <workdir> and enqueues them for processing.
For large batches (>processall_throttle_threshold files), tasks are staggered
to avoid overwhelming downstream APIs.
"""
target_dir = settings.workdir
if not os.path.exists(target_dir):
raise HTTPException(
status_code=400, detail=f"Directory {target_dir} does not exist."
)
raise HTTPException(status_code=400, detail=f"Directory {target_dir} does not exist.")
pdf_files = []
for filename in os.listdir(target_dir):
@@ -128,13 +127,37 @@ def process_all_pdfs_in_workdir():
return {"message": "No PDF files found in that directory."}
task_ids = []
for pdf in pdf_files:
num_files = len(pdf_files)
# Apply throttling if we have more files than the threshold
apply_throttle = num_files > settings.processall_throttle_threshold
if apply_throttle:
logger.info(
f"Processing {num_files} files with throttling "
f"(threshold: {settings.processall_throttle_threshold}, "
f"delay: {settings.processall_throttle_delay}s per file)"
)
for index, pdf in enumerate(pdf_files):
file_path = os.path.join(target_dir, pdf)
task = process_document.delay(file_path)
if apply_throttle:
# Stagger task submission with countdown
# First file starts immediately (countdown=0)
# Each subsequent file has an increasing delay
countdown = index * settings.processall_throttle_delay
task = process_document.apply_async(args=[file_path], countdown=countdown)
logger.debug(f"Scheduled {pdf} with {countdown}s delay")
else:
# No throttling - enqueue immediately
task = process_document.delay(file_path)
task_ids.append(task.id)
return {
"message": f"Enqueued {len(pdf_files)} PDFs to upload_to_s3",
"pdf_files": pdf_files,
"task_ids": task_ids
}
message = f"Enqueued {num_files} PDFs for processing"
if apply_throttle:
total_time = (num_files - 1) * settings.processall_throttle_delay
message += f" (throttled over {total_time} seconds)"
return {"message": message, "pdf_files": pdf_files, "task_ids": task_ids, "throttled": apply_throttle}
app/config.py
+44 -43
View File
@@ -1,10 +1,12 @@
#!/usr/bin/env python3
from pydantic_settings import BaseSettings
from typing import Optional, List, Dict, Any, Union
from pydantic import Field, validator
import os
from datetime import datetime
from typing import Any, Dict, List, Optional, Union
from pydantic import Field, validator
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
@@ -14,23 +16,23 @@ class Settings(BaseSettings):
openai_model: str = "gpt-4o-mini" # Default model
workdir: str
debug: bool = False # Default to False
# Making Dropbox optional
dropbox_app_key: Optional[str] = None
dropbox_app_secret: Optional[str] = None
dropbox_folder: Optional[str] = None
dropbox_refresh_token: Optional[str] = None
# Making Nextcloud optional
nextcloud_upload_url: Optional[str] = None
nextcloud_username: Optional[str] = None
nextcloud_password: Optional[str] = None
nextcloud_folder: Optional[str] = None
# Making Paperless optional
paperless_ngx_api_token: Optional[str] = None
paperless_host: Optional[str] = None
azure_ai_key: str
azure_region: str
azure_endpoint: str
@@ -71,7 +73,7 @@ class Settings(BaseSettings):
google_drive_credentials_json: Optional[str] = ""
google_drive_folder_id: Optional[str] = ""
google_drive_delegate_to: Optional[str] = "" # Optional delegated user email
# Google Drive OAuth settings
google_drive_use_oauth: bool = False # Default to service account method
google_drive_client_id: Optional[str] = ""
@@ -135,49 +137,45 @@ class Settings(BaseSettings):
# Feature flags
allow_file_delete: bool = True # Default to allowing file deletion from database
# Batch processing settings
processall_throttle_threshold: int = Field(
default=20, description="Number of files above which throttling is applied in /processall endpoint"
)
processall_throttle_delay: int = Field(
default=3, description="Delay in seconds between each task submission when throttling in /processall"
)
# Notification settings
notification_urls: Union[List[str], str] = Field(
default_factory=list,
description="List of Apprise notification URLs (e.g., discord://, telegram://, etc.)"
)
notify_on_task_failure: bool = Field(
default=True,
description="Send notifications when Celery tasks fail"
default_factory=list, description="List of Apprise notification URLs (e.g., discord://, telegram://, etc.)"
)
notify_on_task_failure: bool = Field(default=True, description="Send notifications when Celery tasks fail")
notify_on_credential_failure: bool = Field(
default=True,
description="Send notifications when credential checks fail"
)
notify_on_startup: bool = Field(
default=True,
description="Send notifications when application starts"
)
notify_on_shutdown: bool = Field(
default=False,
description="Send notifications when application shuts down"
default=True, description="Send notifications when credential checks fail"
)
notify_on_startup: bool = Field(default=True, description="Send notifications when application starts")
notify_on_shutdown: bool = Field(default=False, description="Send notifications when application shuts down")
notify_on_file_processed: bool = Field(
default=True,
description="Send notifications when files are successfully processed"
default=True, description="Send notifications when files are successfully processed"
)
@validator('notification_urls', pre=True)
@validator("notification_urls", pre=True)
def parse_notification_urls(cls, v):
"""Parse notification URLs from string or list"""
if isinstance(v, str):
if ',' in v:
return [url.strip() for url in v.split(',') if url.strip()]
if "," in v:
return [url.strip() for url in v.split(",") if url.strip()]
elif v.strip():
return [v.strip()]
return []
return v
@validator('session_secret')
@validator("session_secret")
def validate_session_secret(cls, v, values):
"""Validate that session_secret is set and has sufficient length when auth is enabled"""
if values.get('auth_enabled') and not v:
if values.get("auth_enabled") and not v:
raise ValueError("SESSION_SECRET must be set when AUTH_ENABLED=True")
if values.get('auth_enabled') and v and len(v) < 32:
if values.get("auth_enabled") and v and len(v) < 32:
raise ValueError("SESSION_SECRET must be at least 32 characters long")
return v
@@ -188,13 +186,13 @@ class Settings(BaseSettings):
env_build_date = os.environ.get("BUILD_DATE")
if env_build_date:
return env_build_date
# Then try to get build date from BUILD_DATE file
build_date_file = os.path.join(os.path.dirname(os.path.dirname(__file__)), "BUILD_DATE")
if os.path.exists(build_date_file):
with open(build_date_file, "r") as f:
return f.read().strip()
# Default to unknown if not found
return "Unknown build date"
@@ -205,35 +203,38 @@ class Settings(BaseSettings):
env_version = os.environ.get("APP_VERSION")
if env_version:
return env_version
# Then try to get version from VERSION file
version_file = os.path.join(os.path.dirname(os.path.dirname(__file__)), "VERSION")
if os.path.exists(version_file):
with open(version_file, "r") as f:
return f.read().strip()
# Default version if not found
return "0.3.2-dev"
class Config:
env_file = ".env"
# Convert string representations of booleans to actual booleans
# and strip quotes from string values
@classmethod
def parse_env_var(cls, field_name: str, raw_val: str) -> Any:
# First, strip quotes from the value if it's a string
if isinstance(raw_val, str):
if (raw_val.startswith('"') and raw_val.endswith('"')) or \
(raw_val.startswith("'") and raw_val.endswith("'")):
if (raw_val.startswith('"') and raw_val.endswith('"')) or (
raw_val.startswith("'") and raw_val.endswith("'")
):
raw_val = raw_val[1:-1]
raw_val = raw_val.strip()
# Convert string representations of booleans to actual booleans
if field_name.endswith('_enabled') or field_name == 'debug':
if raw_val.lower() in ('false', '0', 'no', 'n', 'f'):
if field_name.endswith("_enabled") or field_name == "debug":
if raw_val.lower() in ("false", "0", "no", "n", "f"):
return False
if raw_val.lower() in ('true', '1', 'yes', 'y', 't'):
if raw_val.lower() in ("true", "1", "yes", "y", "t"):
return True
return raw_val
settings = Settings()
docs/API.md
+23
View File
@@ -101,6 +101,29 @@ Reprocess a specific file.
}
```
### Batch Processing
**POST** `/api/processall`
Process all PDF files in the configured workdir directory.
**Throttling**: For large batches (>20 files by default), tasks are automatically staggered to prevent overwhelming downstream APIs. The throttling behavior can be configured via environment variables:
- `PROCESSALL_THROTTLE_THRESHOLD`: Number of files above which throttling is applied (default: 20)
- `PROCESSALL_THROTTLE_DELAY`: Delay in seconds between each task submission when throttling (default: 3)
**Example**: When processing 25 files with default settings, the first file is queued immediately, the second after 3 seconds, the third after 6 seconds, etc., spreading the load over 72 seconds total.
**Response**:
```json
{
"message": "Enqueued 25 PDFs for processing (throttled over 72 seconds)",
"pdf_files": ["file1.pdf", "file2.pdf", ...],
"task_ids": ["a1b2c3...", "d4e5f6...", ...],
"throttled": true
}
```
**POST** `/send_to_google_drive/`
Send a processed file to Google Drive.
docs/ConfigurationGuide.md
+14
View File
@@ -17,6 +17,20 @@ Configuration is primarily done through environment variables specified in a `.e
| `EXTERNAL_HOSTNAME` | The external hostname for the application. | `docuelevate.example.com` |
| `ALLOW_FILE_DELETE` | Enable file deletion in the web interface (`true`/`false`). | `true` |
### Batch Processing Settings
Control how the `/processall` endpoint handles large batches of files to prevent overwhelming downstream APIs.
| **Variable** | **Description** | **Default** |
|-----------------------------------|----------------------------------------------------------------------------------------------------|-------------|
| `PROCESSALL_THROTTLE_THRESHOLD` | Number of files above which throttling is applied. Files <= threshold are processed immediately. | `20` |
| `PROCESSALL_THROTTLE_DELAY` | Delay in seconds between each task submission when throttling is active. | `3` |
**Example Usage**: When processing 25 files with default settings:
- Files are staggered: file 0 at 0s, file 1 at 3s, file 2 at 6s, etc.
- Total queue time: (25-1) × 3 = 72 seconds
- Prevents API rate limit issues and ensures smooth processing
### IMAP Configuration
DocuElevate can monitor multiple IMAP mailboxes for document attachments. Each mailbox uses a numbered prefix (e.g., `IMAP1_`, `IMAP2_`).
tests/conftest.py
+2 -1
View File
@@ -70,7 +70,8 @@ def client(db_session) -> TestClient:
fastapi_app.dependency_overrides[get_db] = override_get_db
with TestClient(fastapi_app) as test_client:
# Use base_url to satisfy TrustedHostMiddleware
with TestClient(fastapi_app, base_url="http://localhost") as test_client:
yield test_client
# Clean up
tests/test_processall_throttling.py
+241
View File
@@ -0,0 +1,241 @@
"""
Tests for /processall endpoint throttling behavior.
"""
import os
from unittest.mock import MagicMock, Mock, patch
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def mock_throttle_settings(monkeypatch):
"""Fixture to configure throttle settings for tests."""
def _configure(workdir, threshold=20, delay=3):
from app.api import process
monkeypatch.setattr(process.settings, "workdir", str(workdir))
monkeypatch.setattr(process.settings, "processall_throttle_threshold", threshold)
monkeypatch.setattr(process.settings, "processall_throttle_delay", delay)
return _configure
@pytest.mark.integration
class TestProcessAllThrottling:
"""Tests for processall endpoint with throttling."""
@patch("app.api.process.process_document")
def test_processall_no_throttling_for_small_batch(
self, mock_task, client: TestClient, tmp_path, mock_throttle_settings
):
"""Test that small batches (<=20 files) are not throttled."""
# Create test directory with 10 PDF files
workdir = tmp_path / "workdir"
workdir.mkdir()
for i in range(10):
(workdir / f"test{i}.pdf").write_text("dummy pdf content")
# Mock the task
mock_task.delay = Mock(return_value=Mock(id="task-id"))
mock_task.apply_async = Mock(return_value=Mock(id="task-id"))
# Configure settings
mock_throttle_settings(workdir, threshold=20, delay=3)
response = client.post("/api/processall")
assert response.status_code == 200
data = response.json()
# Should use .delay() (not throttled)
assert mock_task.delay.call_count == 10
assert mock_task.apply_async.call_count == 0
# Response should indicate no throttling
assert data["throttled"] is False
assert len(data["pdf_files"]) == 10
assert len(data["task_ids"]) == 10
@patch("app.api.process.process_document")
def test_processall_throttling_for_large_batch(
self, mock_task, client: TestClient, tmp_path, mock_throttle_settings
):
"""Test that large batches (>20 files) are throttled."""
# Create test directory with 25 PDF files
workdir = tmp_path / "workdir"
workdir.mkdir()
for i in range(25):
(workdir / f"test{i}.pdf").write_text("dummy pdf content")
# Mock the task
mock_task_result = Mock(id="task-id")
mock_task.apply_async = Mock(return_value=mock_task_result)
# Configure settings
mock_throttle_settings(workdir, threshold=20, delay=3)
response = client.post("/api/processall")
assert response.status_code == 200
data = response.json()
# Should use .apply_async() with countdown (throttled)
assert mock_task.apply_async.call_count == 25
# Verify countdown values are increasing
calls = mock_task.apply_async.call_args_list
for i, call_args in enumerate(calls):
expected_countdown = i * 3 # 3 seconds delay
assert call_args[1]["countdown"] == expected_countdown
# Response should indicate throttling
assert data["throttled"] is True
assert len(data["pdf_files"]) == 25
assert len(data["task_ids"]) == 25
assert "throttled over" in data["message"]
@patch("app.api.process.process_document")
def test_processall_exactly_at_threshold(self, mock_task, client: TestClient, tmp_path, mock_throttle_settings):
"""Test behavior when file count equals threshold."""
# Create test directory with exactly 20 PDF files
workdir = tmp_path / "workdir"
workdir.mkdir()
for i in range(20):
(workdir / f"test{i}.pdf").write_text("dummy pdf content")
# Mock the task
mock_task.delay = Mock(return_value=Mock(id="task-id"))
# Configure settings
mock_throttle_settings(workdir, threshold=20, delay=3)
response = client.post("/api/processall")
assert response.status_code == 200
data = response.json()
# At threshold, should NOT throttle (only >threshold)
assert mock_task.delay.call_count == 20
assert data["throttled"] is False
@patch("app.api.process.process_document")
def test_processall_one_over_threshold(self, mock_task, client: TestClient, tmp_path, mock_throttle_settings):
"""Test that throttling activates at threshold + 1."""
# Create test directory with 21 PDF files (threshold is 20)
workdir = tmp_path / "workdir"
workdir.mkdir()
for i in range(21):
(workdir / f"test{i}.pdf").write_text("dummy pdf content")
# Mock the task
mock_task.apply_async = Mock(return_value=Mock(id="task-id"))
# Configure settings
mock_throttle_settings(workdir, threshold=20, delay=3)
response = client.post("/api/processall")
assert response.status_code == 200
data = response.json()
# Should be throttled
assert mock_task.apply_async.call_count == 21
assert data["throttled"] is True
def test_processall_empty_directory(self, client: TestClient, tmp_path, mock_throttle_settings):
"""Test processall with no PDF files."""
workdir = tmp_path / "workdir"
workdir.mkdir()
# Configure settings
mock_throttle_settings(workdir)
response = client.post("/api/processall")
assert response.status_code == 200
data = response.json()
assert data["message"] == "No PDF files found in that directory."
def test_processall_nonexistent_directory(self, client: TestClient, tmp_path, mock_throttle_settings):
"""Test processall with non-existent directory."""
workdir = tmp_path / "nonexistent"
# Configure settings
mock_throttle_settings(workdir)
response = client.post("/api/processall")
assert response.status_code == 400
data = response.json()
assert "does not exist" in data["detail"]
@patch("app.api.process.process_document")
def test_processall_custom_threshold(self, mock_task, client: TestClient, tmp_path, mock_throttle_settings):
"""Test that custom threshold value is respected."""
# Create test directory with 15 PDF files
workdir = tmp_path / "workdir"
workdir.mkdir()
for i in range(15):
(workdir / f"test{i}.pdf").write_text("dummy pdf content")
# Mock the task
mock_task.apply_async = Mock(return_value=Mock(id="task-id"))
# Configure settings with lower threshold
mock_throttle_settings(workdir, threshold=10, delay=2)
response = client.post("/api/processall")
assert response.status_code == 200
data = response.json()
# Should be throttled because 15 > 10
assert mock_task.apply_async.call_count == 15
assert data["throttled"] is True
@pytest.mark.unit
class TestThrottlingConfiguration:
"""Tests for throttling configuration settings."""
def test_default_throttle_threshold(self):
"""Test that default threshold is 20."""
from app.config import Settings
settings = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost",
openai_api_key="test-key",
workdir="/tmp",
azure_ai_key="test-key",
azure_region="test-region",
azure_endpoint="https://test.endpoint",
gotenberg_url="http://gotenberg",
session_secret="a" * 32,
)
assert settings.processall_throttle_threshold == 20
def test_default_throttle_delay(self):
"""Test that default delay is 3 seconds."""
from app.config import Settings
settings = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost",
openai_api_key="test-key",
workdir="/tmp",
azure_ai_key="test-key",
azure_region="test-region",
azure_endpoint="https://test.endpoint",
gotenberg_url="http://gotenberg",
session_secret="a" * 32,
)
assert settings.processall_throttle_delay == 3