merge

Author: phact

src/connectors/google_drive_acl.py

@@ -100,10 +100,7 @@ async def _get_cloud_identity_group_roles(
     roles: list[str] = []
     seen: set[str] = set()
     page_token: str | None = None
-    query = (
-        f"member_key_id == '{_cel_string(user_email)}' && "
-        f"'{GOOGLE_GROUP_LABEL}' in labels"
-    )
+    query = f"member_key_id == '{_cel_string(user_email)}' && '{GOOGLE_GROUP_LABEL}' in labels"
 
     try:
         while True:

src/connectors/langflow_connector_service.py

@@ -99,7 +99,7 @@ async def process_connector_document(
             # Step 1: Upload file to Langflow
             logger.debug("Uploading file to Langflow", filename=document.filename)
             content = document.content
-            
+
             # Clean filename and ensure we don't add a double extension
             processed_filename = clean_connector_filename(document.filename, document.mimetype)
 
@@ -114,19 +114,30 @@ async def process_connector_document(
             if self.session_manager:
                 try:
                     from config.settings import get_index_name
-                    opensearch_client = self.session_manager.get_user_opensearch_client(owner_user_id, jwt_token)
+
+                    opensearch_client = self.session_manager.get_user_opensearch_client(
+                        owner_user_id, jwt_token
+                    )
                     delete_body = {"query": {"term": {"filename": processed_filename}}}
-                    delete_result = await opensearch_client.delete_by_query(index=get_index_name(), body=delete_body)
+                    delete_result = await opensearch_client.delete_by_query(
+                        index=get_index_name(), body=delete_body
+                    )
                     deleted_count = delete_result.get("deleted", 0)
-                    logger.info("Deleted existing chunks before re-ingestion", filename=processed_filename, deleted_count=deleted_count)
+                    logger.info(
+                        "Deleted existing chunks before re-ingestion",
+                        filename=processed_filename,
+                        deleted_count=deleted_count,
+                    )
                 except Exception as delete_err:
-                    logger.warning("Failed to delete existing chunks before re-ingestion", filename=processed_filename, error=str(delete_err))
+                    logger.warning(
+                        "Failed to delete existing chunks before re-ingestion",
+                        filename=processed_filename,
+                        error=str(delete_err),
+                    )
 
             langflow_file_id = None  # Initialize to track if upload succeeded
             try:
-                upload_result = await self.langflow_service.upload_user_file(
-                    file_tuple, jwt_token
-                )
+                upload_result = await self.langflow_service.upload_user_file(file_tuple, jwt_token)
                 langflow_file_id = upload_result["id"]
                 langflow_file_path = upload_result["path"]
 
@@ -137,9 +148,7 @@ async def process_connector_document(
                 )
 
                 # Step 2: Run ingestion flow with the uploaded file
-                logger.debug(
-                    "Running Langflow ingestion flow", file_path=langflow_file_path
-                )
+                logger.debug("Running Langflow ingestion flow", file_path=langflow_file_path)
 
                 connector_tweak_settings = None
                 if isinstance(ingest_settings, dict):
@@ -222,7 +231,6 @@ async def process_connector_document(
                         )
                 raise
 
-
     async def sync_connector_files(
         self,
         connection_id: str,
@@ -246,9 +254,7 @@ async def sync_connector_files(
 
         connector = await self.get_connector(connection_id)
         if not connector:
-            raise ValueError(
-                f"Connection '{connection_id}' not found or not authenticated"
-            )
+            raise ValueError(f"Connection '{connection_id}' not found or not authenticated")
 
         logger.debug("Got connector", authenticated=connector.is_authenticated)
 
@@ -264,13 +270,9 @@ async def sync_connector_files(
 
         while True:
             # List files from connector with limit
-            logger.debug(
-                "Calling list_files", page_size=page_size, page_token=page_token
-            )
+            logger.debug("Calling list_files", page_size=page_size, page_token=page_token)
             file_list = await connector.list_files(page_token, limit=page_size)
-            logger.debug(
-                "Got files from connector", file_count=len(file_list.get("files", []))
-            )
+            logger.debug("Got files from connector", file_count=len(file_list.get("files", [])))
             files = file_list["files"]
 
             if not files:
@@ -333,7 +335,7 @@ async def sync_specific_files(
         """
         Sync specific files by their IDs using Langflow processing.
         Automatically expands folders to their contents.
-        
+
         Args:
             connection_id: The connection ID
             user_id: The user ID
@@ -351,9 +353,7 @@ async def sync_specific_files(
 
         connector = await self.get_connector(connection_id)
         if not connector:
-            raise ValueError(
-                f"Connection '{connection_id}' not found or not authenticated"
-            )
+            raise ValueError(f"Connection '{connection_id}' not found or not authenticated")
 
         if not connector.is_authenticated:
             raise ValueError(f"Connection '{connection_id}' not authenticated")
@@ -368,7 +368,7 @@ async def sync_specific_files(
 
         # If file_infos provided, cache them in the connector for later use
         # This allows get_file_content to use download URLs directly
-        if file_infos and hasattr(connector, 'set_file_infos'):
+        if file_infos and hasattr(connector, "set_file_infos"):
             connector.set_file_infos(file_infos)
             logger.info(f"Cached {len(file_infos)} file infos with download URLs in connector")
 
@@ -435,9 +435,7 @@ async def sync_specific_files(
         original_filenames = {}
         if file_infos:
             original_filenames = {
-                f["id"]: clean_connector_filename(
-                    f["name"], f.get("mimeType") or f.get("mimetype")
-                )
+                f["id"]: clean_connector_filename(f["name"], f.get("mimeType") or f.get("mimetype"))
                 for f in file_infos
                 if "id" in f and "name" in f
             }

src/connectors/onedrive/connector.py

@@ -488,9 +488,7 @@ async def _extract_onedrive_acl(self, file_id: str, file_metadata: Dict) -> Docu
 
                 # Granted to identities (can include users and groups)
                 identities = (
-                    perm.get("grantedToIdentitiesV2")
-                    or perm.get("grantedToIdentities")
-                    or []
+                    perm.get("grantedToIdentitiesV2") or perm.get("grantedToIdentities") or []
                 )
                 if identities:
                     for identity in identities:

src/session_manager.py

@@ -254,7 +254,6 @@ def create_opensearch_jwt_token(
         """Create a short-lived OpenSearch JWT with current connector groups."""
         if ttl_seconds is None:
             from config.settings import get_opensearch_jwt_ttl_seconds
-
             ttl_seconds = get_opensearch_jwt_ttl_seconds()
         return self._create_signed_jwt_token(
             user,