diff --git a/.env.demo b/.env.demo index 7aaa0e90..6f22583a 100644 --- a/.env.demo +++ b/.env.demo @@ -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: diff --git a/app/api/process.py b/app/api/process.py index 8163a7fe..e99f7d82 100644 --- a/app/api/process.py +++ b/app/api/process.py @@ -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 and enqueues them for processing.""" + """ + Finds all .pdf files in 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} diff --git a/app/config.py b/app/config.py index b723b74e..b9e7fc96 100644 --- a/app/config.py +++ b/app/config.py @@ -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() diff --git a/docs/API.md b/docs/API.md index f113731b..45636a13 100644 --- a/docs/API.md +++ b/docs/API.md @@ -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. diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md index 04fc0edb..0aef4ee6 100644 --- a/docs/ConfigurationGuide.md +++ b/docs/ConfigurationGuide.md @@ -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_`). diff --git a/tests/conftest.py b/tests/conftest.py index 4e80504d..2b949b50 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 diff --git a/tests/test_processall_throttling.py b/tests/test_processall_throttling.py new file mode 100644 index 00000000..5b2d65b8 --- /dev/null +++ b/tests/test_processall_throttling.py @@ -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