fix: resolve merge conflict in database.py

Merge origin/main into branch, resolving conflict in app/database.py.
Combined improvements from both branches:
- Keep pool_pre_ping=True and structured variable approach from feature branch
- Add explicit QueuePool import and poolclass assignment from main
This commit is contained in:
copilot-swe-agent[bot]
2026-03-19 10:40:43 +00:00
55 changed files with 2924 additions and 109 deletions
+68 -11
View File
@@ -13,7 +13,7 @@ plaintext is returned exactly once at creation time.
import hashlib
import logging
import secrets
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request, status
@@ -105,6 +105,7 @@ def _token_to_dict(t: ApiToken) -> dict[str, Any]:
"last_used_ip": t.last_used_ip,
"created_at": t.created_at,
"revoked_at": t.revoked_at,
"expires_at": t.expires_at,
}
@@ -117,6 +118,12 @@ class TokenCreate(BaseModel):
"""Schema for creating a new API token."""
name: str = Field(..., min_length=1, max_length=255, description="Human-readable label for the token")
expires_in_days: int | None = Field(
default=None,
ge=1,
le=3650, # Maximum 10 years; keeps tokens from being effectively permanent while allowing long-lived CI/CD tokens.
description="Optional lifetime in days. If omitted the token never expires.",
)
class TokenResponse(BaseModel):
@@ -130,6 +137,7 @@ class TokenResponse(BaseModel):
last_used_ip: str | None
created_at: datetime | None
revoked_at: datetime | None
expires_at: datetime | None
model_config = {"from_attributes": True}
@@ -160,11 +168,16 @@ async def create_token(
token_hash_value = hash_token(plaintext)
prefix = plaintext[:12] # "de_" prefix + 9 random chars = 12 chars total
expires_at = None
if body.expires_in_days is not None:
expires_at = datetime.now(timezone.utc) + timedelta(days=body.expires_in_days)
db_token = ApiToken(
owner_id=owner_id,
name=body.name,
token_hash=token_hash_value,
token_prefix=prefix,
expires_at=expires_at,
)
try:
db.add(db_token)
@@ -185,6 +198,7 @@ async def create_token(
"last_used_ip": db_token.last_used_ip,
"created_at": db_token.created_at,
"revoked_at": db_token.revoked_at,
"expires_at": db_token.expires_at,
"token": plaintext,
}
@@ -235,30 +249,73 @@ async def list_mobile_tokens(
@router.delete("/{token_id}", status_code=status.HTTP_200_OK)
async def revoke_token(
async def revoke_or_delete_token(
token_id: int,
owner_id: CurrentOwner,
db: DbSession,
) -> dict[str, str]:
"""Revoke (soft-delete) an API token.
"""Revoke or permanently delete an API token.
The token row is kept for audit purposes but marked inactive with a
``revoked_at`` timestamp.
* **Active token** – soft-revoked: the row is kept for audit purposes
but marked inactive with a ``revoked_at`` timestamp.
* **Already-revoked token** – hard-deleted: the row is permanently
removed from the database.
"""
db_token = db.query(ApiToken).filter(ApiToken.id == token_id, ApiToken.owner_id == owner_id).first()
if not db_token:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Token not found")
if not db_token.is_active:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Token is already revoked")
if db_token.is_active:
# Soft-revoke the active token.
try:
db_token.is_active = False
db_token.revoked_at = datetime.now(timezone.utc)
db.commit()
except Exception:
db.rollback()
raise
logger.info("API token revoked: id=%s owner=%s", token_id, owner_id)
return {"detail": "Token revoked"}
# Hard-delete an already-revoked token.
try:
db_token.is_active = False
db_token.revoked_at = datetime.now(timezone.utc)
db.delete(db_token)
db.commit()
except Exception:
db.rollback()
raise
logger.info("API token permanently deleted: id=%s owner=%s", token_id, owner_id)
return {"detail": "Token deleted"}
logger.info("API token revoked: id=%s owner=%s", token_id, owner_id)
return {"detail": "Token revoked"}
@router.post("/{token_id}/reactivate", status_code=status.HTTP_200_OK, response_model=TokenResponse)
async def reactivate_token(
token_id: int,
owner_id: CurrentOwner,
db: DbSession,
) -> dict[str, Any]:
"""Reactivate a previously revoked API token.
Clears the ``revoked_at`` timestamp and sets ``is_active`` back to
``True``. The token can be used for authentication again immediately.
If the token had an ``expires_at`` in the past the caller should
consider re-creating a new token instead.
"""
db_token = db.query(ApiToken).filter(ApiToken.id == token_id, ApiToken.owner_id == owner_id).first()
if not db_token:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Token not found")
if db_token.is_active:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Token is already active")
try:
db_token.is_active = True
db_token.revoked_at = None
db.commit()
db.refresh(db_token)
except Exception:
db.rollback()
raise
logger.info("API token reactivated: id=%s owner=%s", token_id, owner_id)
return _token_to_dict(db_token)
+27 -9
View File
@@ -347,7 +347,9 @@ def bulk_delete_files(request: Request, file_ids: List[int], db: DbSession):
try:
# Find all file records
file_records = db.query(FileRecord).filter(FileRecord.id.in_(file_ids)).all()
query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
query = apply_owner_filter(query, request)
file_records = query.all()
if not file_records:
raise HTTPException(status_code=404, detail="No files found with the provided IDs")
@@ -386,7 +388,9 @@ def bulk_reprocess_files(request: Request, file_ids: List[int], db: DbSession):
"""
try:
# Find all file records
file_records = db.query(FileRecord).filter(FileRecord.id.in_(file_ids)).all()
query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
query = apply_owner_filter(query, request)
file_records = query.all()
if not file_records:
raise HTTPException(status_code=404, detail="No files found with the provided IDs")
@@ -458,7 +462,9 @@ def bulk_reprocess_files_cloud_ocr(request: Request, file_ids: List[int], db: Db
Useful for re-running OCR on files with poor text quality or missing OCR text.
"""
try:
file_records = db.query(FileRecord).filter(FileRecord.id.in_(file_ids)).all()
query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
query = apply_owner_filter(query, request)
file_records = query.all()
if not file_records:
raise HTTPException(status_code=404, detail="No files found with the provided IDs")
@@ -538,7 +544,9 @@ def bulk_download_files(request: Request, file_ids: List[int], db: DbSession):
Files not found on disk are silently skipped.
"""
try:
file_records = db.query(FileRecord).filter(FileRecord.id.in_(file_ids)).all()
query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
query = apply_owner_filter(query, request)
file_records = query.all()
if not file_records:
raise HTTPException(status_code=404, detail="No files found with the provided IDs")
@@ -620,7 +628,9 @@ def reprocess_single_file(request: Request, file_id: int, db: DbSession):
"""
try:
# Find the file record
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
query = db.query(FileRecord).filter(FileRecord.id == file_id)
query = apply_owner_filter(query, request)
file_record = query.first()
if not file_record:
raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
@@ -676,7 +686,9 @@ def reprocess_with_cloud_ocr(request: Request, file_id: int, db: DbSession):
"""
try:
# Find the file record
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
query = db.query(FileRecord).filter(FileRecord.id == file_id)
query = apply_owner_filter(query, request)
file_record = query.first()
if not file_record:
raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
@@ -939,7 +951,9 @@ def retry_subtask(
"""
try:
# Find the file record
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
query = db.query(FileRecord).filter(FileRecord.id == file_id)
query = apply_owner_filter(query, request)
file_record = query.first()
if not file_record:
raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
@@ -1081,7 +1095,9 @@ def get_file_preview(
try:
# Find the file record
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
query = db.query(FileRecord).filter(FileRecord.id == file_id)
query = apply_owner_filter(query, request)
file_record = query.first()
if not file_record:
raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
@@ -1161,7 +1177,9 @@ def download_file(
try:
# Find the file record
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
query = db.query(FileRecord).filter(FileRecord.id == file_id)
query = apply_owner_filter(query, request)
file_record = query.first()
if not file_record:
raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")
+21 -8
View File
@@ -273,31 +273,44 @@ async def list_devices(
return [_device_to_response(d) for d in devices]
@router.delete("/devices/{device_id}", status_code=status.HTTP_204_NO_CONTENT)
@router.delete("/devices/{device_id}", status_code=status.HTTP_200_OK)
@require_login
async def deactivate_device(
request: Request,
device_id: int,
owner_id: CurrentOwner,
db: DbSession,
) -> None:
"""Deactivate a push-notification device registration.
) -> dict[str, str]:
"""Deactivate or permanently delete a push-notification device registration.
The device record is kept for audit purposes but will no longer receive
push notifications.
* **Active device** – soft-deactivated: the record is kept for audit
purposes but will no longer receive push notifications.
* **Already-inactive device** – hard-deleted: the record is permanently
removed from the database.
"""
device = db.get(MobileDevice, device_id)
if not device or device.owner_id != owner_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Device not found")
device.is_active = False
if device.is_active:
device.is_active = False
try:
db.commit()
except Exception:
db.rollback()
raise
logger.info("Mobile device deactivated: id=%s owner=%s", device_id, owner_id)
return {"detail": "Device deactivated"}
# Hard-delete an already-inactive device.
try:
db.delete(device)
db.commit()
except Exception:
db.rollback()
raise
logger.info("Mobile device deactivated: id=%s owner=%s", device_id, owner_id)
logger.info("Mobile device permanently deleted: id=%s owner=%s", device_id, owner_id)
return {"detail": "Device deleted"}
@router.get("/whoami", response_model=WhoAmIResponse)
+35
View File
@@ -19,10 +19,13 @@ Security properties:
from __future__ import annotations
import base64
import io
import logging
from datetime import datetime
from typing import Annotated, Any
import segno
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
@@ -70,7 +73,9 @@ class CreateChallengeResponse(BaseModel):
challenge_id: int
challenge_token: str
expires_at: datetime
ttl_seconds: int = Field(description="Seconds until the challenge expires (use for client-side countdown).")
qr_payload: str = Field(description="The string to encode in the QR code.")
qr_code_svg: str = Field(description="Base64-encoded SVG data URI of the QR code, ready for use in an <img> src.")
class ChallengeStatusResponse(BaseModel):
@@ -105,6 +110,29 @@ class ClaimChallengeResponse(BaseModel):
created_at: datetime
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
# QR code rendering parameters
_QR_ERROR_LEVEL = "M" # Medium error correction (~15% recovery); sufficient for on-screen display
_QR_SCALE = 4 # Each QR module is rendered as 4×4 SVG pixels
def _generate_qr_svg(payload: str) -> str:
"""Generate a QR code for *payload* and return it as a base64 SVG data URI.
Using ``segno`` (pure-Python, no Pillow dependency) and SVG output so the
QR code scales crisply at any resolution without requiring a canvas or any
client-side JavaScript library.
"""
qr = segno.make(payload, error=_QR_ERROR_LEVEL)
buf = io.BytesIO()
qr.save(buf, kind="svg", scale=_QR_SCALE, xmldecl=False, svgclass=None, lineclass=None, omitsize=True)
svg_bytes = buf.getvalue()
return "data:image/svg+xml;base64," + base64.b64encode(svg_bytes).decode("ascii")
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@@ -131,11 +159,18 @@ async def create_challenge(
base_url = str(request.base_url).rstrip("/")
qr_payload = f"docuelevate://qr-login?token={challenge.challenge_token}&server={base_url}"
# Compute the TTL in seconds so the client can run a countdown timer
# without comparing absolute timestamps (which breaks when client and
# server clocks are out of sync).
ttl_seconds = max(0, int((challenge.expires_at - challenge.created_at).total_seconds()))
return {
"challenge_id": challenge.id,
"challenge_token": challenge.challenge_token,
"expires_at": challenge.expires_at,
"ttl_seconds": ttl_seconds,
"qr_payload": qr_payload,
"qr_code_svg": _generate_qr_svg(qr_payload),
}
+10
View File
@@ -186,6 +186,16 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
logger.debug("[AUTH] _resolve_bearer_user: no active API token matched the provided hash")
return None
# Reject tokens that have passed their optional expiry.
if db_token.expires_at is not None:
now_utc = datetime.now(timezone.utc)
expires_aware = db_token.expires_at
if expires_aware.tzinfo is None:
expires_aware = expires_aware.replace(tzinfo=timezone.utc)
if now_utc > expires_aware:
logger.debug("[AUTH] _resolve_bearer_user: API token id=%s has expired", db_token.id)
return None
logger.debug(
"[AUTH] _resolve_bearer_user: matched API token id=%s owner=%s",
db_token.id,
+1
View File
@@ -52,6 +52,7 @@ from app.tasks.upload_to_onedrive import upload_to_onedrive # noqa: F401
from app.tasks.upload_to_paperless import upload_to_paperless # noqa: F401
from app.tasks.upload_to_s3 import upload_to_s3 # noqa: F401
from app.tasks.upload_to_sftp import upload_to_sftp # noqa: F401
from app.tasks.upload_to_sharepoint import upload_to_sharepoint # noqa: F401
from app.tasks.upload_to_user_integration import upload_to_user_integration # noqa: F401
from app.tasks.upload_to_webdav import upload_to_webdav # noqa: F401
from app.tasks.upload_with_rclone import send_to_all_rclone_destinations, upload_with_rclone # noqa: F401
+48
View File
@@ -627,6 +627,15 @@ class Settings(BaseSettings):
onedrive_refresh_token: Optional[str] = None # Required for personal accounts
onedrive_folder_path: Optional[str] = None
# SharePoint settings
sharepoint_client_id: Optional[str] = None
sharepoint_client_secret: Optional[str] = None
sharepoint_tenant_id: Optional[str] = "common"
sharepoint_refresh_token: Optional[str] = None
sharepoint_site_url: Optional[str] = None # e.g. https://tenant.sharepoint.com/sites/sitename
sharepoint_document_library: Optional[str] = "Documents" # Document library name
sharepoint_folder_path: Optional[str] = None # Subfolder inside the library
# AWS S3 settings
s3_enabled: bool = Field(
default=True,
@@ -1124,6 +1133,45 @@ class Settings(BaseSettings):
),
)
# Database Connection Pool Configuration
# Controls SQLAlchemy QueuePool behaviour for PostgreSQL/MySQL.
# SQLite uses NullPool and ignores these settings.
db_pool_size: int = Field(
default=5,
description="Number of persistent connections kept in the pool. Ignored for SQLite.",
)
db_max_overflow: int = Field(
default=10,
description=("Maximum number of connections that can be opened beyond db_pool_size. Ignored for SQLite."),
)
db_pool_timeout: int = Field(
default=30,
description="Seconds to wait for a connection from the pool before raising an error. Ignored for SQLite.",
)
db_pool_recycle: int = Field(
default=1800,
description=(
"Seconds after which a connection is recycled to prevent stale connections. "
"Ignored for SQLite. Default: 1800 (30 minutes)."
),
)
# Per-user upload rate limiting (health-aware limiter)
# Controls how many uploads a single user may submit within a sliding window.
upload_rate_limit_per_user: int = Field(
default=20,
description=(
"Maximum number of uploads allowed per user within the upload_rate_limit_window. "
"The limiter may dynamically reduce this value when Redis queue depth or CPU load is high."
),
)
upload_rate_limit_window: int = Field(
default=60,
description=(
"Sliding window in seconds over which upload_rate_limit_per_user is enforced. Default: 60 seconds."
),
)
# Rate Limiting Configuration (see SECURITY_AUDIT.md and docs/API.md)
# Protects against DoS attacks and API abuse
rate_limiting_enabled: bool = Field(
+2 -1
View File
@@ -10,7 +10,7 @@ from typing import Any
from sqlalchemy import create_engine, exc
from sqlalchemy.engine.url import make_url
from sqlalchemy.orm import Session, declarative_base, sessionmaker
from sqlalchemy.pool import NullPool
from sqlalchemy.pool import NullPool, QueuePool
from app.config import settings
@@ -38,6 +38,7 @@ if _parsed_url.get_backend_name() == "sqlite":
_engine_kwargs["poolclass"] = NullPool
else:
# PostgreSQL / MySQL — use a bounded QueuePool with configurable limits.
_engine_kwargs["poolclass"] = QueuePool
_engine_kwargs.update(
{
"pool_size": settings.db_pool_size,
+7
View File
@@ -20,6 +20,9 @@ How it works:
Exempt paths (CSRF is not checked even for state-changing methods):
- ``/oauth-callback`` – OAuth 2.0 callback; protected by the ``state`` parameter.
- ``/api/qr-auth/claim`` – Called by the unauthenticated mobile app; the
cryptographically-random, single-use challenge token provides equivalent
protection.
"""
import logging
@@ -39,6 +42,10 @@ CSRF_PROTECTED_METHODS = {"POST", "PUT", "DELETE", "PATCH"}
# their own replay-protection mechanism).
CSRF_EXEMPT_PATHS = {
"/oauth-callback",
# The mobile app calls this endpoint without a browser session/CSRF token.
# The cryptographically-random, single-use challenge token already provides
# equivalent protection against cross-site request forgery.
"/api/qr-auth/claim",
}
+5
View File
@@ -603,6 +603,7 @@ class IntegrationType:
EMAIL = "EMAIL"
PAPERLESS = "PAPERLESS"
RCLONE = "RCLONE"
SHAREPOINT = "SHAREPOINT"
ICLOUD = "ICLOUD"
ALL = {
@@ -620,6 +621,7 @@ class IntegrationType:
EMAIL,
PAPERLESS,
RCLONE,
SHAREPOINT,
ICLOUD,
}
@@ -784,6 +786,9 @@ class ApiToken(Base):
created_at = Column(DateTime(timezone=True), server_default=func.now())
revoked_at = Column(DateTime(timezone=True), nullable=True)
# Optional expiry: if set, the token is rejected after this timestamp.
expires_at = Column(DateTime(timezone=True), nullable=True)
class SharedLink(Base):
"""Shareable, time-limited or view-limited document link.
+19
View File
@@ -18,6 +18,7 @@ from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.upload_to_paperless import upload_to_paperless
from app.tasks.upload_to_s3 import upload_to_s3
from app.tasks.upload_to_sftp import upload_to_sftp
from app.tasks.upload_to_sharepoint import upload_to_sharepoint
from app.tasks.upload_to_webdav import upload_to_webdav
from app.utils.config_validator import get_provider_status
from app.utils.logging import log_task_progress
@@ -121,6 +122,18 @@ def _should_upload_to_icloud():
return bool(getattr(settings, "icloud_enabled", True) and settings.icloud_username and settings.icloud_password)
def _should_upload_to_sharepoint():
return bool(
settings.sharepoint_client_id
and settings.sharepoint_client_secret
and settings.sharepoint_site_url
and (
settings.sharepoint_refresh_token
or (settings.sharepoint_tenant_id and settings.sharepoint_tenant_id != "common")
)
)
def get_configured_services_from_validator():
"""
Use the config validator to determine which services are configured and enabled.
@@ -140,6 +153,7 @@ def get_configured_services_from_validator():
"Email": "email",
"OneDrive": "onedrive",
"S3 Storage": "s3",
"SharePoint": "sharepoint",
"iCloud Drive": "icloud",
}
@@ -250,6 +264,11 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id:
"should_upload": _should_upload_to_s3,
"upload_func": upload_to_s3,
},
{
"name": "sharepoint",
"should_upload": _should_upload_to_sharepoint,
"upload_func": upload_to_sharepoint,
},
{
"name": "icloud",
"should_upload": _should_upload_to_icloud,
+338
View File
@@ -0,0 +1,338 @@
#!/usr/bin/env python3
"""Upload documents to Microsoft SharePoint via the Microsoft Graph API.
This module authenticates using MSAL (same OAuth2 flow as OneDrive) and
uploads files to a configurable SharePoint Online document library using
the chunked upload session approach for reliability with large files.
Key differences from the OneDrive provider:
- Uses ``/sites/{siteId}/drives/{driveId}`` instead of ``/me/drive``
- Requires a SharePoint site URL to resolve the site and drive IDs
- Targets a named document library (default: ``Documents``)
"""
import logging
import os
import time
import urllib.parse
import msal
import requests
from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import UploadTaskWithRetry
from app.utils import log_task_progress
logger = logging.getLogger(__name__)
def get_sharepoint_token() -> str:
"""Acquire a Microsoft Graph API access token for SharePoint.
Uses MSAL ``ConfidentialClientApplication`` with the refresh-token flow
(delegated permissions) or the client-credentials flow (application
permissions) depending on configuration.
Returns:
A valid access token string.
Raises:
ValueError: When required settings are missing or token acquisition fails.
"""
if not settings.sharepoint_client_id or not settings.sharepoint_client_secret:
raise ValueError("SharePoint client ID and client secret must be configured")
tenant = settings.sharepoint_tenant_id or "common"
logger.info("Using SharePoint tenant: %s", tenant)
scopes = ["https://graph.microsoft.com/.default"]
if settings.sharepoint_refresh_token:
app = msal.ConfidentialClientApplication(
client_id=settings.sharepoint_client_id,
client_credential=settings.sharepoint_client_secret,
authority=f"https://login.microsoftonline.com/{tenant}",
)
logger.info("Attempting to acquire SharePoint token using refresh token")
token_response = app.acquire_token_by_refresh_token(
refresh_token=settings.sharepoint_refresh_token, scopes=scopes
)
if "access_token" not in token_response:
error = token_response.get("error", "")
error_desc = token_response.get("error_description", "Unknown error")
logger.error("Failed to get SharePoint access token: %s - %s", error, error_desc)
raise ValueError(f"Failed to get SharePoint access token: {error} - {error_desc}")
if "refresh_token" in token_response:
settings.sharepoint_refresh_token = token_response["refresh_token"]
logger.info("Updated SharePoint refresh token in memory")
return token_response["access_token"]
elif settings.sharepoint_tenant_id and settings.sharepoint_tenant_id != "common":
authority = f"https://login.microsoftonline.com/{settings.sharepoint_tenant_id}"
app = msal.ConfidentialClientApplication(
client_id=settings.sharepoint_client_id,
client_credential=settings.sharepoint_client_secret,
authority=authority,
)
token_response = app.acquire_token_for_client(scopes=scopes)
if "access_token" not in token_response:
error = token_response.get("error", "")
error_desc = token_response.get("error_description", "Unknown error")
raise ValueError(f"Failed to get SharePoint access token: {error} - {error_desc}")
return token_response["access_token"]
else:
raise ValueError("For SharePoint, either a refresh token or a non-'common' tenant ID is required")
def resolve_sharepoint_drive(access_token: str, site_url: str, library_name: str) -> tuple[str, str]:
"""Resolve the Graph API site ID and drive ID for a SharePoint site.
Args:
access_token: Valid Microsoft Graph API token.
site_url: Full SharePoint site URL, e.g.
``https://tenant.sharepoint.com/sites/sitename``.
library_name: Display name of the document library (e.g. ``Documents``).
Returns:
A ``(site_id, drive_id)`` tuple.
Raises:
ValueError: When the site URL cannot be parsed.
RuntimeError: When the Graph API call fails.
"""
parsed = urllib.parse.urlparse(site_url)
hostname = parsed.hostname
site_path = parsed.path.rstrip("/")
if not hostname or not site_path:
raise ValueError(
f"Invalid SharePoint site URL '{site_url}'. Expected format: https://tenant.sharepoint.com/sites/sitename"
)
headers = {"Authorization": f"Bearer {access_token}"}
# Resolve site ID
site_api_url = f"https://graph.microsoft.com/v1.0/sites/{hostname}:{site_path}"
logger.info("Resolving SharePoint site: %s", site_api_url)
resp = requests.get(site_api_url, headers=headers, timeout=settings.http_request_timeout)
if resp.status_code != 200:
raise RuntimeError(f"Failed to resolve SharePoint site: {resp.status_code} - {resp.text}")
site_id = resp.json()["id"]
logger.info("Resolved SharePoint site ID: %s", site_id)
# Resolve drive ID from the document library name
drives_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives"
resp = requests.get(drives_url, headers=headers, timeout=settings.http_request_timeout)
if resp.status_code != 200:
raise RuntimeError(f"Failed to list SharePoint drives: {resp.status_code} - {resp.text}")
drives = resp.json().get("value", [])
drive_id = None
for drive in drives:
if drive.get("name", "").lower() == library_name.lower():
drive_id = drive["id"]
break
if not drive_id:
available = [d.get("name") for d in drives]
raise RuntimeError(f"Document library '{library_name}' not found on site. Available libraries: {available}")
logger.info("Resolved SharePoint drive ID: %s (library: %s)", drive_id, library_name)
return site_id, drive_id
def create_sharepoint_upload_session(
filename: str, folder_path: str | None, drive_id: str, site_id: str, access_token: str
) -> str:
"""Create a resumable upload session on a SharePoint document library.
Args:
filename: Name of the file to upload.
folder_path: Optional subfolder path inside the library.
drive_id: Graph API drive ID of the document library.
site_id: Graph API site ID.
access_token: Valid access token.
Returns:
The upload session URL for chunked PUT requests.
Raises:
RuntimeError: When session creation fails.
"""
base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
if folder_path:
folder_path = folder_path.strip("/")
path_components = folder_path.split("/")
encoded_path = "/".join(urllib.parse.quote(component) for component in path_components)
encoded_filename = urllib.parse.quote(filename)
item_path = f"/root:/{encoded_path}/{encoded_filename}:/createUploadSession"
else:
encoded_filename = urllib.parse.quote(filename)
item_path = f"/root:/{encoded_filename}:/createUploadSession"
url = f"{base_url}{item_path}"
request_body = {"item": {"@microsoft.graph.conflictBehavior": "replace"}}
headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
logger.info("Creating SharePoint upload session for %s at path %s", filename, folder_path)
response = requests.post(url, headers=headers, json=request_body, timeout=settings.http_request_timeout)
if response.status_code == 200:
upload_url = response.json().get("uploadUrl")
logger.info("SharePoint upload session created for %s", filename)
return upload_url
else:
raise RuntimeError(f"Failed to create SharePoint upload session: {response.status_code} - {response.text}")
def upload_large_file_sharepoint(file_path: str, upload_url: str) -> dict:
"""Upload a file to SharePoint using a chunked upload session.
Args:
file_path: Local path to the file.
upload_url: The upload session URL from ``create_sharepoint_upload_session``.
Returns:
The Graph API response dict containing file metadata.
Raises:
RuntimeError: When a chunk upload fails after retries.
"""
file_size = os.path.getsize(file_path)
chunk_size = 10 * 1024 * 1024 # 10 MB
response = None
with open(file_path, "rb") as f:
chunk_number = 0
while True:
chunk = f.read(chunk_size)
if not chunk:
break
chunk_start = chunk_number * chunk_size
chunk_end = chunk_start + len(chunk) - 1
content_range = f"bytes {chunk_start}-{chunk_end}/{file_size}"
headers = {"Content-Length": str(len(chunk)), "Content-Range": content_range}
max_retries = 3
retry_delay = 2
for attempt in range(max_retries):
try:
response = requests.put(
upload_url, headers=headers, data=chunk, timeout=settings.http_request_timeout
)
if response.status_code in (201, 202):
break
else:
logger.warning(
"SharePoint chunk upload failed (attempt %d): %d", attempt + 1, response.status_code
)
if attempt < max_retries - 1:
time.sleep(retry_delay * (attempt + 1))
except Exception as e:
logger.warning("SharePoint chunk upload error (attempt %d): %s", attempt + 1, str(e))
if attempt < max_retries - 1:
time.sleep(retry_delay * (attempt + 1))
if response is None or response.status_code not in (201, 202):
status = response.status_code if response else "no response"
text = response.text if response else ""
raise RuntimeError(f"Failed to upload chunk after {max_retries} attempts: {status} - {text}")
chunk_number += 1
return response.json() if response else {}
@celery.task(base=UploadTaskWithRetry, bind=True)
def upload_to_sharepoint(self, file_path: str, file_id: int = None, folder_override: str = None):
"""Upload a file to SharePoint Online.
Args:
file_path: Path to the file to upload.
file_id: Optional file ID to associate with logs.
folder_override: Optional folder path override.
Returns:
A dict with upload status and file details.
Raises:
FileNotFoundError: When the file does not exist.
ValueError: When SharePoint is not configured.
RuntimeError: When the upload fails.
"""
task_id = self.request.id
logger.info("[%s] Starting SharePoint upload: %s", task_id, file_path)
log_task_progress(
task_id,
"upload_to_sharepoint",
"in_progress",
f"Uploading to SharePoint: {os.path.basename(file_path)}",
file_id=file_id,
)
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}"
logger.error("[%s] %s", task_id, error_msg)
log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
filename = os.path.basename(file_path)
if not settings.sharepoint_client_id:
error_msg = "SharePoint client ID is not configured"
logger.error("[%s] %s", task_id, error_msg)
log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
raise ValueError(error_msg)
if not settings.sharepoint_site_url:
error_msg = "SharePoint site URL is not configured"
logger.error("[%s] %s", task_id, error_msg)
log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
raise ValueError(error_msg)
try:
access_token = get_sharepoint_token()
library_name = settings.sharepoint_document_library or "Documents"
site_id, drive_id = resolve_sharepoint_drive(access_token, settings.sharepoint_site_url, library_name)
folder_path = folder_override if folder_override is not None else settings.sharepoint_folder_path
upload_url = create_sharepoint_upload_session(filename, folder_path, drive_id, site_id, access_token)
result = upload_large_file_sharepoint(file_path, upload_url)
web_url = result.get("webUrl", "Not available")
logger.info("[%s] Successfully uploaded %s to SharePoint", task_id, filename)
logger.info("[%s] File accessible at: %s", task_id, web_url)
log_task_progress(
task_id, "upload_to_sharepoint", "success", f"Uploaded to SharePoint: {filename}", file_id=file_id
)
return {
"status": "Completed",
"file_path": file_path,
"sharepoint_path": f"{folder_path or ''}/{filename}",
"web_url": web_url,
}
except Exception as e:
error_msg = f"Failed to upload {filename} to SharePoint: {str(e)}"
logger.error("[%s] %s", task_id, error_msg)
log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
raise RuntimeError(error_msg) from e
+108
View File
@@ -571,6 +571,113 @@ def _upload_rclone(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], t
return {"status": "Completed", "rclone_dest": dest}
def _upload_sharepoint(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], task_id: str) -> dict[str, Any]:
"""Upload *file_path* to SharePoint using per-user MSAL credentials."""
import urllib.parse
import msal
import requests as _requests
client_id = creds.get("client_id") or ""
client_secret = creds.get("client_secret") or ""
refresh_token = creds.get("refresh_token") or ""
tenant = cfg.get("tenant_id") or "common"
site_url = cfg.get("site_url") or ""
library_name = cfg.get("document_library") or "Documents"
folder_path = cfg.get("folder_path") or ""
if not (client_id and client_secret):
raise ValueError("SharePoint integration is missing client_id or client_secret in credentials")
if not site_url:
raise ValueError("SharePoint integration is missing site_url in config")
scopes = ["https://graph.microsoft.com/.default"]
msal_app = msal.ConfidentialClientApplication(
client_id=client_id,
client_credential=client_secret,
authority=f"https://login.microsoftonline.com/{tenant}",
)
if refresh_token:
token_resp = msal_app.acquire_token_by_refresh_token(refresh_token=refresh_token, scopes=scopes)
else:
token_resp = msal_app.acquire_token_for_client(scopes=scopes)
if "access_token" not in token_resp:
raise ValueError(f"SharePoint token acquisition failed: {token_resp.get('error_description', 'unknown')}")
access_token = token_resp["access_token"]
headers = {"Authorization": f"Bearer {access_token}"}
# Resolve site ID
parsed = urllib.parse.urlparse(site_url)
hostname = parsed.hostname
site_path = parsed.path.rstrip("/")
if not hostname or not site_path:
raise ValueError(f"Invalid SharePoint site URL: {site_url}")
resp = _requests.get(f"https://graph.microsoft.com/v1.0/sites/{hostname}:{site_path}", headers=headers, timeout=30)
resp.raise_for_status()
site_id = resp.json()["id"]
# Resolve drive ID
resp = _requests.get(f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives", headers=headers, timeout=30)
resp.raise_for_status()
drive_id = None
for drive in resp.json().get("value", []):
if drive.get("name", "").lower() == library_name.lower():
drive_id = drive["id"]
break
if not drive_id:
raise RuntimeError(f"Document library '{library_name}' not found on SharePoint site")
filename = os.path.basename(file_path)
# Build upload-session URL
base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
if folder_path:
folder_path = folder_path.strip("/")
encoded_path = "/".join(urllib.parse.quote(p) for p in folder_path.split("/"))
encoded_file = urllib.parse.quote(filename)
item_path = f"/root:/{encoded_path}/{encoded_file}:/createUploadSession"
else:
encoded_file = urllib.parse.quote(filename)
item_path = f"/root:/{encoded_file}:/createUploadSession"
session_url = f"{base_url}{item_path}"
session_headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
resp = _requests.post(
session_url,
headers=session_headers,
json={"item": {"@microsoft.graph.conflictBehavior": "replace"}},
timeout=30,
)
resp.raise_for_status()
upload_url = resp.json()["uploadUrl"]
file_size = os.path.getsize(file_path)
chunk_size = 10 * 1024 * 1024
with open(file_path, "rb") as fh:
chunk_num = 0
while True:
chunk = fh.read(chunk_size)
if not chunk:
break
start = chunk_num * chunk_size
end = start + len(chunk) - 1
upload_headers = {
"Content-Length": str(len(chunk)),
"Content-Range": f"bytes {start}-{end}/{file_size}",
}
upload_resp = _requests.put(upload_url, headers=upload_headers, data=chunk, timeout=120)
if upload_resp.status_code not in (201, 202):
raise RuntimeError(f"SharePoint chunk upload failed: {upload_resp.status_code}")
chunk_num += 1
logger.info("[%s] SharePoint upload complete: %s/%s", task_id, folder_path, filename)
return {"status": "Completed", "sharepoint_folder": folder_path, "filename": filename}
def _upload_icloud(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], task_id: str) -> dict[str, Any]:
"""Upload *file_path* to iCloud Drive using per-user credentials.
@@ -615,6 +722,7 @@ _UPLOAD_HANDLERS = {
IntegrationType.PAPERLESS: _upload_paperless,
IntegrationType.EMAIL: _upload_email,
IntegrationType.RCLONE: _upload_rclone,
IntegrationType.SHAREPOINT: _upload_sharepoint,
IntegrationType.ICLOUD: _upload_icloud,
}
+22
View File
@@ -296,6 +296,28 @@ def get_provider_status() -> dict[str, dict[str, object]]:
},
}
# Check SharePoint configuration
providers["SharePoint"] = {
"name": "SharePoint",
"icon": "fa-brands fa-microsoft",
"configured": bool(
getattr(settings, "sharepoint_client_id", None)
and getattr(settings, "sharepoint_client_secret", None)
and getattr(settings, "sharepoint_site_url", None)
),
"enabled": True,
"description": "Store documents in Microsoft SharePoint Online",
"details": {
"client_id": getattr(settings, "sharepoint_client_id", "Not set"),
"client_secret": mask_sensitive_value(getattr(settings, "sharepoint_client_secret", None)),
"tenant_id": getattr(settings, "sharepoint_tenant_id", "Not set"),
"refresh_token": mask_sensitive_value(getattr(settings, "sharepoint_refresh_token", None)),
"site_url": getattr(settings, "sharepoint_site_url", "Not set"),
"document_library": getattr(settings, "sharepoint_document_library", "Not set"),
"folder_path": getattr(settings, "sharepoint_folder_path", "Not set"),
},
}
# Check S3 configuration
providers["S3 Storage"] = {
"name": "S3 Storage",
+137 -2
View File
@@ -943,6 +943,63 @@ SETTING_METADATA = {
"required": False,
"restart_required": False,
},
# Storage Providers - SharePoint
"sharepoint_client_id": {
"category": "Storage Providers",
"description": "SharePoint Azure AD application (client) ID",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"sharepoint_client_secret": {
"category": "Storage Providers",
"description": "SharePoint Azure AD client secret",
"type": "string",
"sensitive": True,
"required": False,
"restart_required": False,
},
"sharepoint_tenant_id": {
"category": "Storage Providers",
"description": "SharePoint Azure AD tenant ID (use 'common' for multi-tenant apps)",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"sharepoint_refresh_token": {
"category": "Storage Providers",
"description": "SharePoint OAuth refresh token",
"type": "string",
"sensitive": True,
"required": False,
"restart_required": False,
},
"sharepoint_site_url": {
"category": "Storage Providers",
"description": "SharePoint site URL (e.g. https://tenant.sharepoint.com/sites/sitename)",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"sharepoint_document_library": {
"category": "Storage Providers",
"description": "SharePoint document library name (default: 'Documents')",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"sharepoint_folder_path": {
"category": "Storage Providers",
"description": "Subfolder path inside the SharePoint document library",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Storage Providers - WebDAV
"webdav_enabled": {
"category": "Storage Providers",
@@ -2013,14 +2070,26 @@ SETTING_METADATA = {
"category": "Backup",
"description": (
"Storage provider for remote backup copies. "
"Accepted values: s3, dropbox, google_drive, onedrive, nextcloud, webdav, ftp, sftp, email. "
"Accepted values: s3, dropbox, google_drive, onedrive, sharepoint, nextcloud, webdav, ftp, sftp, email. "
"Leave empty to keep backups local only."
),
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
"options": ["", "s3", "dropbox", "google_drive", "onedrive", "nextcloud", "webdav", "ftp", "sftp", "email"],
"options": [
"",
"s3",
"dropbox",
"google_drive",
"onedrive",
"sharepoint",
"nextcloud",
"webdav",
"ftp",
"sftp",
"email",
],
},
"backup_remote_folder": {
"category": "Backup",
@@ -2532,6 +2601,72 @@ SETTING_METADATA = {
"required": False,
"restart_required": False,
},
# Database Connection Pool
"db_pool_size": {
"category": "Core",
"description": (
"Number of persistent connections kept in the SQLAlchemy QueuePool. "
"Has no effect for SQLite databases. Default: 5."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_max_overflow": {
"category": "Core",
"description": (
"Maximum extra connections that can be opened beyond db_pool_size. "
"Has no effect for SQLite databases. Default: 10."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_pool_timeout": {
"category": "Core",
"description": (
"Seconds to wait for a connection from the pool before raising an error. "
"Has no effect for SQLite databases. Default: 30."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"db_pool_recycle": {
"category": "Core",
"description": (
"Seconds after which idle connections are recycled to prevent stale connections. "
"Has no effect for SQLite databases. Default: 1800 (30 minutes)."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
# Per-user upload rate limiting
"upload_rate_limit_per_user": {
"category": "Security",
"description": (
"Maximum number of uploads a single user may submit within upload_rate_limit_window seconds. "
"The health-aware limiter may reduce this dynamically under high Redis queue depth or CPU load. "
"Default: 20."
),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": False,
},
"upload_rate_limit_window": {
"category": "Security",
"description": ("Sliding window in seconds over which upload_rate_limit_per_user is enforced. Default: 60."),
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Rate Limiting
"rate_limiting_enabled": {
"category": "Security",