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

feat(imap): add per-user IMAP ingestion accounts with quota enforcement

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-08 13:01:14 +00:00
parent f3c85f73c3
commit 43bcfe5653
12 changed files with 1922 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/__init__.py
+2
View File
@@ -16,6 +16,7 @@ from app.api.dropbox import router as dropbox_router
from app.api.duplicates import router as duplicates_router
from app.api.files import router as files_router
from app.api.google_drive import router as google_drive_router
from app.api.imap_accounts import router as imap_accounts_router
from app.api.logs import router as logs_router
from app.api.onboarding import router as onboarding_router
from app.api.onedrive import router as onedrive_router
@@ -68,3 +69,4 @@ router.include_router(plans_router)
router.include_router(onboarding_router)
router.include_router(billing_router)
router.include_router(pipelines_router)
router.include_router(imap_accounts_router)
app/api/imap_accounts.py
+350
View File
@@ -0,0 +1,350 @@
"""API endpoints for managing per-user IMAP ingestion accounts.
Provides CRUD operations for a user's IMAP accounts, quota enforcement
against their subscription plan's ``max_mailboxes`` limit, and a
test-connection endpoint so users can verify credentials before saving.
"""
import imaplib
import logging
from datetime import datetime, timezone
from typing import Annotated, Any
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import UserImapAccount
from app.utils.subscription import get_tier, get_user_tier_id
from app.utils.user_scope import get_current_owner_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/imap-accounts", tags=["imap-accounts"])
DbSession = Annotated[Session, Depends(get_db)]
# ---------------------------------------------------------------------------
# Auth helpers
# ---------------------------------------------------------------------------
def _get_owner_id(request: Request) -> str:
"""Return the current user's owner ID, raising 401 if unauthenticated."""
owner_id = get_current_owner_id(request)
if owner_id is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
return owner_id
CurrentOwner = Annotated[str, Depends(_get_owner_id)]
# ---------------------------------------------------------------------------
# Quota helpers
# ---------------------------------------------------------------------------
_FREE_TIER_ID = "free"
def _get_max_mailboxes(tier: dict[str, Any]) -> int | None:
"""Return the maximum number of IMAP accounts allowed by *tier*.
Returns:
``None`` — unlimited (paid tiers with ``max_mailboxes == 0``)
``0`` — no mailboxes allowed (free tier)
positive — the configured limit
"""
tier_id: str = tier.get("id", _FREE_TIER_ID)
max_mb: int = tier.get("max_mailboxes", 0)
# Free tier: 0 means "no access" (not "unlimited")
if tier_id == _FREE_TIER_ID:
return 0
# Paid tiers: 0 means unlimited
if max_mb == 0:
return None
return max_mb
def _check_quota(db: Session, owner_id: str) -> None:
"""Raise 403 if the user has reached their IMAP account quota."""
tier_id = get_user_tier_id(db, owner_id)
tier = get_tier(tier_id, db)
max_mb = _get_max_mailboxes(tier)
if max_mb == 0:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=("Your current plan does not include email ingestion. Upgrade to a paid plan to add IMAP accounts."),
)
if max_mb is not None:
current_count = db.query(UserImapAccount).filter(UserImapAccount.owner_id == owner_id).count()
if current_count >= max_mb:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=(
f"You have reached your plan limit of {max_mb} IMAP account(s). "
"Please delete an existing account or upgrade your plan."
),
)
# ---------------------------------------------------------------------------
# Pydantic schemas
# ---------------------------------------------------------------------------
class ImapAccountCreate(BaseModel):
"""Schema for creating a new IMAP account."""
name: str = Field(..., min_length=1, max_length=255, description="Human-readable label")
host: str = Field(..., min_length=1, max_length=255, description="IMAP server hostname")
port: int = Field(default=993, ge=1, le=65535, description="IMAP server port")
username: str = Field(..., min_length=1, max_length=255, description="IMAP login username")
password: str = Field(..., min_length=1, max_length=1024, description="IMAP login password")
use_ssl: bool = Field(default=True, description="Use SSL/TLS connection")
delete_after_process: bool = Field(default=False, description="Delete emails from mailbox after processing")
is_active: bool = Field(default=True, description="Whether to poll this mailbox")
class ImapAccountUpdate(BaseModel):
"""Schema for updating an existing IMAP account (all fields optional)."""
name: str | None = Field(default=None, min_length=1, max_length=255)
host: str | None = Field(default=None, min_length=1, max_length=255)
port: int | None = Field(default=None, ge=1, le=65535)
username: str | None = Field(default=None, min_length=1, max_length=255)
password: str | None = Field(default=None, min_length=1, max_length=1024)
use_ssl: bool | None = None
delete_after_process: bool | None = None
is_active: bool | None = None
class ImapTestRequest(BaseModel):
"""Schema for testing an IMAP connection without saving it."""
host: str = Field(..., min_length=1, max_length=255)
port: int = Field(default=993, ge=1, le=65535)
username: str = Field(..., min_length=1, max_length=255)
password: str = Field(..., min_length=1, max_length=1024)
use_ssl: bool = Field(default=True)
# ---------------------------------------------------------------------------
# Serialisation helpers
# ---------------------------------------------------------------------------
def _to_response(acct: UserImapAccount) -> dict[str, Any]:
"""Serialize a ``UserImapAccount`` row to a response dict.
Passwords are never included in responses.
"""
return {
"id": acct.id,
"owner_id": acct.owner_id,
"name": acct.name,
"host": acct.host,
"port": acct.port,
"username": acct.username,
"use_ssl": acct.use_ssl,
"delete_after_process": acct.delete_after_process,
"is_active": acct.is_active,
"last_checked_at": acct.last_checked_at.isoformat() if acct.last_checked_at else None,
"last_error": acct.last_error,
"created_at": acct.created_at.isoformat() if acct.created_at else None,
"updated_at": acct.updated_at.isoformat() if acct.updated_at else None,
}
# ---------------------------------------------------------------------------
# Connection test helper
# ---------------------------------------------------------------------------
def _test_imap_connection(host: str, port: int, username: str, password: str, use_ssl: bool) -> dict[str, Any]:
"""Attempt to connect and log in to the IMAP server.
Returns a dict with ``{"success": bool, "message": str}``.
"""
try:
if use_ssl:
mail = imaplib.IMAP4_SSL(host, port)
else:
mail = imaplib.IMAP4(host, port)
mail.login(username, password)
mail.logout()
return {"success": True, "message": "Connection successful"}
except OSError as exc:
logger.warning("IMAP network error for %s@%s: %s", username, host, exc)
return {"success": False, "message": f"Connection error: {exc}"}
except Exception as exc: # noqa: BLE001
logger.warning("IMAP error for %s@%s: %s", username, host, exc)
return {"success": False, "message": f"IMAP error: {exc}"}
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@router.get("/", summary="List IMAP accounts for the current user")
def list_imap_accounts(request: Request, db: DbSession, owner_id: CurrentOwner) -> list[dict[str, Any]]:
"""Return all IMAP accounts belonging to the authenticated user."""
accounts = db.query(UserImapAccount).filter(UserImapAccount.owner_id == owner_id).order_by(UserImapAccount.id).all()
return [_to_response(a) for a in accounts]
@router.post("/", status_code=status.HTTP_201_CREATED, summary="Create a new IMAP account")
def create_imap_account(
request: Request, body: ImapAccountCreate, db: DbSession, owner_id: CurrentOwner
) -> dict[str, Any]:
"""Create a new IMAP ingestion account for the current user.
Quota is enforced against the user's subscription plan's ``max_mailboxes``
limit before the account is persisted.
"""
_check_quota(db, owner_id)
acct = UserImapAccount(
owner_id=owner_id,
name=body.name,
host=body.host,
port=body.port,
username=body.username,
password=body.password,
use_ssl=body.use_ssl,
delete_after_process=body.delete_after_process,
is_active=body.is_active,
)
try:
db.add(acct)
db.commit()
db.refresh(acct)
except Exception:
db.rollback()
raise
logger.info("User %s created IMAP account %d (%s)", owner_id, acct.id, body.host)
return _to_response(acct)
@router.get("/{account_id}", summary="Get a single IMAP account")
def get_imap_account(account_id: int, request: Request, db: DbSession, owner_id: CurrentOwner) -> dict[str, Any]:
"""Return a single IMAP account by ID (must belong to the current user)."""
acct = (
db.query(UserImapAccount).filter(UserImapAccount.id == account_id, UserImapAccount.owner_id == owner_id).first()
)
if not acct:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="IMAP account not found")
return _to_response(acct)
@router.put("/{account_id}", summary="Update an IMAP account")
def update_imap_account(
account_id: int,
request: Request,
body: ImapAccountUpdate,
db: DbSession,
owner_id: CurrentOwner,
) -> dict[str, Any]:
"""Update an existing IMAP account. Only provided fields are changed."""
acct = (
db.query(UserImapAccount).filter(UserImapAccount.id == account_id, UserImapAccount.owner_id == owner_id).first()
)
if not acct:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="IMAP account not found")
if body.name is not None:
acct.name = body.name
if body.host is not None:
acct.host = body.host
if body.port is not None:
acct.port = body.port
if body.username is not None:
acct.username = body.username
if body.password is not None:
acct.password = body.password
if body.use_ssl is not None:
acct.use_ssl = body.use_ssl
if body.delete_after_process is not None:
acct.delete_after_process = body.delete_after_process
if body.is_active is not None:
acct.is_active = body.is_active
# Reset last_error so the next poll gives a fresh result
acct.last_error = None
acct.updated_at = datetime.now(timezone.utc)
try:
db.commit()
db.refresh(acct)
except Exception:
db.rollback()
raise
logger.info("User %s updated IMAP account %d", owner_id, account_id)
return _to_response(acct)
@router.delete("/{account_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete an IMAP account")
def delete_imap_account(account_id: int, request: Request, db: DbSession, owner_id: CurrentOwner) -> None:
"""Delete an IMAP account permanently."""
acct = (
db.query(UserImapAccount).filter(UserImapAccount.id == account_id, UserImapAccount.owner_id == owner_id).first()
)
if not acct:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="IMAP account not found")
try:
db.delete(acct)
db.commit()
except Exception:
db.rollback()
raise
logger.info("User %s deleted IMAP account %d", owner_id, account_id)
@router.post("/{account_id}/test", summary="Test an existing IMAP account's connection")
def test_saved_imap_account(account_id: int, request: Request, db: DbSession, owner_id: CurrentOwner) -> dict[str, Any]:
"""Test the connection for an already-saved IMAP account."""
acct = (
db.query(UserImapAccount).filter(UserImapAccount.id == account_id, UserImapAccount.owner_id == owner_id).first()
)
if not acct:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="IMAP account not found")
return _test_imap_connection(acct.host, acct.port, acct.username, acct.password, acct.use_ssl)
@router.post("/test", summary="Test an IMAP connection without saving")
def test_imap_connection(request: Request, body: ImapTestRequest, owner_id: CurrentOwner) -> dict[str, Any]:
"""Test IMAP credentials without persisting anything.
Useful for the "Test connection" button in the UI before the user saves
a new account.
"""
return _test_imap_connection(body.host, body.port, body.username, body.password, body.use_ssl)
@router.get("/quota/", summary="Get IMAP account quota information for the current user")
def get_imap_quota(request: Request, db: DbSession, owner_id: CurrentOwner) -> dict[str, Any]:
"""Return the user's current IMAP account usage vs. their plan quota."""
tier_id = get_user_tier_id(db, owner_id)
tier = get_tier(tier_id, db)
max_mb = _get_max_mailboxes(tier)
current_count = db.query(UserImapAccount).filter(UserImapAccount.owner_id == owner_id).count()
return {
"current_count": current_count,
"max_mailboxes": max_mb, # None = unlimited, 0 = not allowed
"can_add": max_mb is None or (max_mb > 0 and current_count < max_mb),
"tier_id": tier_id,
"tier_name": tier.get("name", tier_id),
}