westfarn
57a2350b8e
## Summary - **#59** — Persist and expose Drive sync progress (`sync_total`, `sync_processed`, `sync_added`, `sync_updated`, `sync_failed`) during sync so the FE can show a progress bar. - **#60** — Paginate/search/sort `GET /documents/` with `workspace=personal|company` scope isolation. - **#94 (API)** — Add `GET /analytics/user_prompt_heatmap/?tz=` weekday × hour bins for user-entered prompts. ## Test plan - [x] `manage.py test` drive sync / documents / analytics / drive_tasks suites - [ ] Manual: trigger Drive sync and poll connections → progress fields update while pending - [ ] Manual: `GET /documents/?workspace=personal&page=1&search=…&ordering=name` - [ ] Deploy FE companion PR for #92/#93/#94 UI Closes #59 Closes #60Reviewed-on: #61
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""Drive sync background tasks (#57).
|
|
|
|
Uses Django 6 Tasks API (same pattern as ``email_tasks``). With the default
|
|
``ImmediateBackend``, enqueue still runs in-process — we dispatch that call on
|
|
a daemon thread after commit so ``POST .../sync/`` can return 202 Pending
|
|
without waiting for Drive list/download/ingest. Swap ``TASKS`` to a durable
|
|
queue + worker later; enqueue path stays the same.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from functools import partial
|
|
|
|
from django.conf import settings
|
|
from django.db import transaction
|
|
from django.tasks import task
|
|
|
|
from chat_backend.models import DriveConnection
|
|
from chat_backend.services.drive_sync import sync_connection
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _uses_immediate_backend() -> bool:
|
|
backend = (
|
|
(getattr(settings, "TASKS", {}) or {})
|
|
.get("default", {})
|
|
.get("BACKEND", "")
|
|
)
|
|
return "ImmediateBackend" in str(backend)
|
|
|
|
|
|
@task
|
|
def run_drive_connection_sync(connection_id: int) -> dict:
|
|
"""Load a DriveConnection and run ``sync_connection`` (#57)."""
|
|
try:
|
|
connection = DriveConnection.objects.get(pk=connection_id, is_active=True)
|
|
except DriveConnection.DoesNotExist:
|
|
logger.error(
|
|
"DriveConnection %s missing or inactive; cannot sync", connection_id
|
|
)
|
|
return {"error": "connection_not_found"}
|
|
return sync_connection(connection)
|
|
|
|
|
|
def _dispatch_sync(connection_id: int) -> None:
|
|
"""Enqueue (or run) the sync task; ImmediateBackend runs off the request thread."""
|
|
try:
|
|
if _uses_immediate_backend():
|
|
# ImmediateBackend executes during enqueue — keep HTTP snappy.
|
|
threading.Thread(
|
|
target=run_drive_connection_sync.enqueue,
|
|
kwargs={"connection_id": connection_id},
|
|
daemon=True,
|
|
name=f"drive-sync-{connection_id}",
|
|
).start()
|
|
else:
|
|
run_drive_connection_sync.enqueue(connection_id=connection_id)
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to dispatch Drive sync for connection=%s", connection_id
|
|
)
|
|
DriveConnection.objects.filter(pk=connection_id).update(
|
|
last_sync_status=DriveConnection.SyncStatus.ERROR,
|
|
last_sync_error="Failed to enqueue Drive sync task.",
|
|
)
|
|
|
|
|
|
def enqueue_drive_sync(
|
|
connection: DriveConnection, *, force: bool = False
|
|
) -> tuple[DriveConnection, bool]:
|
|
"""Mark connection pending and enqueue sync after DB commit.
|
|
|
|
Returns ``(connection, enqueued)``. If already ``pending`` and ``force`` is
|
|
false, does not enqueue a duplicate job.
|
|
"""
|
|
connection.refresh_from_db()
|
|
if (
|
|
not force
|
|
and connection.last_sync_status == DriveConnection.SyncStatus.PENDING
|
|
):
|
|
return connection, False
|
|
|
|
connection.last_sync_status = DriveConnection.SyncStatus.PENDING
|
|
connection.last_sync_error = ""
|
|
connection.sync_total = 0
|
|
connection.sync_processed = 0
|
|
connection.sync_added = 0
|
|
connection.sync_updated = 0
|
|
connection.sync_failed = 0
|
|
connection.save(
|
|
update_fields=[
|
|
"last_sync_status",
|
|
"last_sync_error",
|
|
"sync_total",
|
|
"sync_processed",
|
|
"sync_added",
|
|
"sync_updated",
|
|
"sync_failed",
|
|
"last_modified",
|
|
]
|
|
)
|
|
|
|
transaction.on_commit(
|
|
partial(_dispatch_sync, connection_id=connection.id)
|
|
)
|
|
return connection, True
|