Heavy task improvements, logging, and validation (#4058)

backend/ee/onyx/db/connector_credential_pair.py

@@ -4,6 +4,7 @@ from sqlalchemy.orm import Session
 from onyx.configs.constants import DocumentSource
 from onyx.db.connector_credential_pair import get_connector_credential_pair
 from onyx.db.enums import AccessType
+from onyx.db.enums import ConnectorCredentialPairStatus
 from onyx.db.models import Connector
 from onyx.db.models import ConnectorCredentialPair
 from onyx.db.models import UserGroup__ConnectorCredentialPair
@@ -35,10 +36,11 @@ def _delete_connector_credential_pair_user_groups_relationship__no_commit(
 def get_cc_pairs_by_source(
     db_session: Session,
     source_type: DocumentSource,
-    only_sync: bool,
+    access_type: AccessType | None = None,
+    status: ConnectorCredentialPairStatus | None = None,
 ) -> list[ConnectorCredentialPair]:
     """
-    Get all cc_pairs for a given source type (and optionally only sync)
+    Get all cc_pairs for a given source type with optional filtering by access_type and status
     result is sorted by cc_pair id
     """
     query = (
@@ -48,8 +50,11 @@ def get_cc_pairs_by_source(
         .order_by(ConnectorCredentialPair.id)
     )
 
-    if only_sync:
-        query = query.filter(ConnectorCredentialPair.access_type == AccessType.SYNC)
+    if access_type is not None:
+        query = query.filter(ConnectorCredentialPair.access_type == access_type)
+
+    if status is not None:
+        query = query.filter(ConnectorCredentialPair.status == status)
 
     cc_pairs = query.all()
     return cc_pairs

backend/ee/onyx/external_permissions/google_drive/doc_sync.py

@@ -62,12 +62,14 @@ def _fetch_permissions_for_permission_ids(
         user_email=(owner_email or google_drive_connector.primary_admin_email),
     )
 
+    # We continue on 404 or 403 because the document may not exist or the user may not have access to it
     fetched_permissions = execute_paginated_retrieval(
         retrieval_function=drive_service.permissions().list,
         list_key="permissions",
         fileId=doc_id,
         fields="permissions(id, emailAddress, type, domain)",
         supportsAllDrives=True,
+        continue_on_404_or_403=True,
     )
 
     permissions_for_doc_id = []
@@ -104,7 +106,13 @@ def _get_permissions_from_slim_doc(
     user_emails: set[str] = set()
     group_emails: set[str] = set()
     public = False
+    skipped_permissions = 0
+
     for permission in permissions_list:
+        if not permission:
+            skipped_permissions += 1
+            continue
+
         permission_type = permission["type"]
         if permission_type == "user":
             user_emails.add(permission["emailAddress"])
@@ -121,6 +129,11 @@ def _get_permissions_from_slim_doc(
         elif permission_type == "anyone":
             public = True
 
+    if skipped_permissions > 0:
+        logger.warning(
+            f"Skipped {skipped_permissions} permissions of {len(permissions_list)} for document {slim_doc.id}"
+        )
+
     drive_id = permission_info.get("drive_id")
     group_ids = group_emails | ({drive_id} if drive_id is not None else set())