From 8fb3998d67e48f207b3682e5509130b133bb2484 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Feb 2026 14:53:04 +0000 Subject: [PATCH 1/5] Initial plan From 8eea420bc510d4c1792b87ad065db57c0571b65f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Feb 2026 14:59:19 +0000 Subject: [PATCH 2/5] Add throttling to /processall endpoint to prevent overwhelming downstream APIs Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- .env.demo | 5 + app/api/process.py | 44 +++++- app/config.py | 10 ++ tests/conftest.py | 3 +- tests/test_processall_throttling.py | 230 ++++++++++++++++++++++++++++ 5 files changed, 286 insertions(+), 6 deletions(-) create mode 100644 tests/test_processall_throttling.py 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..6e13c004 100644 --- a/app/api/process.py +++ b/app/api/process.py @@ -112,7 +112,12 @@ def send_to_all_destinations_endpoint(file_path: str): @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( @@ -128,13 +133,42 @@ 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) + 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": f"Enqueued {len(pdf_files)} PDFs to upload_to_s3", + "message": message, "pdf_files": pdf_files, - "task_ids": task_ids + "task_ids": task_ids, + "throttled": apply_throttle } diff --git a/app/config.py b/app/config.py index b723b74e..1d579f36 100644 --- a/app/config.py +++ b/app/config.py @@ -135,6 +135,16 @@ 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, 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..6bddd665 --- /dev/null +++ b/tests/test_processall_throttling.py @@ -0,0 +1,230 @@ +""" +Tests for /processall endpoint throttling behavior. +""" +import os +import pytest +from unittest.mock import Mock, patch, MagicMock +from fastapi.testclient import TestClient + + +@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, monkeypatch): + """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")) + + # Use monkeypatch to modify the settings imported in the process module + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(workdir)) + monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) + monkeypatch.setattr(process.settings, 'processall_throttle_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, monkeypatch): + """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) + + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(workdir)) + monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) + monkeypatch.setattr(process.settings, 'processall_throttle_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, monkeypatch): + """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")) + + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(workdir)) + monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) + monkeypatch.setattr(process.settings, 'processall_throttle_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, monkeypatch): + """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")) + + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(workdir)) + monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) + monkeypatch.setattr(process.settings, 'processall_throttle_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, monkeypatch): + """Test processall with no PDF files.""" + workdir = tmp_path / "workdir" + workdir.mkdir() + + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(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, monkeypatch): + """Test processall with non-existent directory.""" + workdir = tmp_path / "nonexistent" + + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(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, monkeypatch): + """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")) + + from app.api import process + monkeypatch.setattr(process.settings, 'workdir', str(workdir)) + monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 10) # Lower threshold + monkeypatch.setattr(process.settings, 'processall_throttle_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 From a12ccdabacf7c96cc78d1a98543e8764fe7936b5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Feb 2026 15:00:18 +0000 Subject: [PATCH 3/5] Apply Black and isort formatting to modified files Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/process.py | 89 ++++++++++++++++++++-------------------------- app/config.py | 85 ++++++++++++++++++++----------------------- 2 files changed, 77 insertions(+), 97 deletions(-) diff --git a/app/api/process.py b/app/api/process.py index 6e13c004..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,101 +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. - + 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): @@ -134,20 +128,20 @@ def process_all_pdfs_in_workdir(): task_ids = [] 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) - + if apply_throttle: # Stagger task submission with countdown # First file starts immediately (countdown=0) @@ -158,17 +152,12 @@ def process_all_pdfs_in_workdir(): else: # No throttling - enqueue immediately task = process_document.delay(file_path) - + task_ids.append(task.id) 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 - } + + 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 1d579f36..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] = "" @@ -137,57 +139,43 @@ class Settings(BaseSettings): # Batch processing settings processall_throttle_threshold: int = Field( - default=20, - description="Number of files above which throttling is applied in /processall endpoint" + 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" + 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 @@ -198,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" @@ -215,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() From 237c9504ad84ca4bdc6dd901e3650218352a4ceb Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Feb 2026 15:00:56 +0000 Subject: [PATCH 4/5] Add documentation for batch processing throttling feature Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- docs/API.md | 23 +++++++++++++++++++++++ docs/ConfigurationGuide.md | 14 ++++++++++++++ 2 files changed, 37 insertions(+) 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_`). From 0de31fc7754c51882b413063322e9dc902e153f2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Feb 2026 15:03:17 +0000 Subject: [PATCH 5/5] Refactor tests to use fixture for better maintainability Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- tests/test_processall_throttling.py | 167 +++++++++++++++------------- 1 file changed, 89 insertions(+), 78 deletions(-) diff --git a/tests/test_processall_throttling.py b/tests/test_processall_throttling.py index 6bddd665..5b2d65b8 100644 --- a/tests/test_processall_throttling.py +++ b/tests/test_processall_throttling.py @@ -1,193 +1,202 @@ """ Tests for /processall endpoint throttling behavior. """ + import os +from unittest.mock import MagicMock, Mock, patch + import pytest -from unittest.mock import Mock, patch, MagicMock 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, monkeypatch): + @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")) - - # Use monkeypatch to modify the settings imported in the process module - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) - monkeypatch.setattr(process.settings, 'processall_throttle_delay', 3) - + + # 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, monkeypatch): + @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) - - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) - monkeypatch.setattr(process.settings, 'processall_throttle_delay', 3) - + + # 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 - + 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, monkeypatch): + @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")) - - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) - monkeypatch.setattr(process.settings, 'processall_throttle_delay', 3) - + + # 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, monkeypatch): + @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")) - - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 20) - monkeypatch.setattr(process.settings, 'processall_throttle_delay', 3) - + + # 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, monkeypatch): + 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() - - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - + + # 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, monkeypatch): + def test_processall_nonexistent_directory(self, client: TestClient, tmp_path, mock_throttle_settings): """Test processall with non-existent directory.""" workdir = tmp_path / "nonexistent" - - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - + + # 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, monkeypatch): + @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")) - - from app.api import process - monkeypatch.setattr(process.settings, 'workdir', str(workdir)) - monkeypatch.setattr(process.settings, 'processall_throttle_threshold', 10) # Lower threshold - monkeypatch.setattr(process.settings, 'processall_throttle_delay', 2) - + + # 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 @@ -200,6 +209,7 @@ class TestThrottlingConfiguration: 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", @@ -216,6 +226,7 @@ class TestThrottlingConfiguration: 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",