Merge remote-tracking branch 'origin/deploy/konwledge' into claude/guards-controllers-244f13

api/.ruff.toml

@@ -70,7 +70,6 @@ ignore = [
     "N815",    # mixed-case-variable-in-class-scope
     "PT011",   # pytest-raises-too-broad
     "SIM102",  # collapsible-if
-    "SIM105",  # suppressible-exception
     "SIM117",  # multiple-with-statements
 ]
 

api/controllers/openapi/_errors.py

@@ -20,6 +20,7 @@ The transport-generic enum members, ``_CODE_BY_STATUS`` and the
 promote them to ``libs`` if a second surface adopts ``ErrorBody``.
 """
 
+import contextlib
 import logging
 from enum import StrEnum
 from typing import Any
@@ -177,10 +178,8 @@ class OpenApiErrorFormatter:
         # flask-restx Api.handle_error does `data = getattr(e, "data", default_data)`
         # AFTER our handler returns, so a pre-existing e.data (flask_restx.abort,
         # BaseHTTPException) would override the canonical body. Rewrite it.
-        try:
+        with contextlib.suppress(AttributeError):
             e.data = wire  # type: ignore[attr-defined]
-        except AttributeError:
-            pass
         return wire
 
     def _resolve_code(self, e: Exception, status_code: int) -> str:

api/core/agent/cot_agent_runner.py

@@ -1,3 +1,4 @@
+import contextlib
 import json
 import logging
 from abc import ABC, abstractmethod
@@ -333,10 +334,8 @@ class CotAgentRunner(BaseAgentRunner, ABC):
             return answer, ToolInvokeMeta.error_instance(answer)
 
         if isinstance(tool_call_args, str):
-            try:
+            with contextlib.suppress(json.JSONDecodeError):
                 tool_call_args = json.loads(tool_call_args)
-            except json.JSONDecodeError:
-                pass
 
         # invoke tool
         with use_credit_usage_metadata({"app_type": CreditUsageAppType.AGENT}):

api/core/app/app_config/workflow_ui_based_app/variables/manager.py

@@ -1,3 +1,4 @@
+import contextlib
 import json
 import re
 from typing import Any
@@ -42,11 +43,9 @@ class WorkflowVariablesConfigManager:
         """
         json_schema = variable.get("json_schema")
         if isinstance(json_schema, str):
-            try:
+            # Leave as-is; Pydantic validation will surface the error.
+            with contextlib.suppress(json.JSONDecodeError, TypeError):
                 variable["json_schema"] = json.loads(json_schema)
-            except (json.JSONDecodeError, TypeError):
-                # Leave as-is; Pydantic validation will surface the error.
-                pass
 
     @classmethod
     def convert_rag_pipeline_variable(cls, workflow: Workflow, start_node_id: str) -> list[RagPipelineVariableEntity]:

api/core/mcp/client/sse_client.py

@@ -2,7 +2,7 @@ import logging
 import queue
 from collections.abc import Generator
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from typing import Any, final
 from urllib.parse import urljoin, urlparse
 
@@ -310,10 +310,8 @@ def sse_client(
     finally:
         # Close the SSE connection to unblock the reader thread
         if transport.event_source is not None:
-            try:
+            with suppress(RuntimeError):
                 transport.event_source.response.close()
-            except RuntimeError:
-                pass
 
         # Clean up queues
         if read_queue:

api/core/provider_manager.py

@@ -1661,13 +1661,11 @@ class ProviderManager:
 
                     for variable in provider_credential_secret_variables:
                         if variable in provider_credentials:
-                            try:
+                            with contextlib.suppress(ValueError):
                                 provider_credentials[variable] = encrypter.decrypt_token_with_decoding(
                                     provider_credentials.get(variable, ""),
                                     decoding_context,
                                 )
-                            except ValueError:
-                                pass
 
                     current_using_credentials = provider_credentials or {}
 
@@ -1803,13 +1801,11 @@ class ProviderManager:
 
                             for variable in model_credential_secret_variables:
                                 if variable in provider_model_credentials:
-                                    try:
+                                    with contextlib.suppress(ValueError):
                                         provider_model_credentials[variable] = encrypter.decrypt_token_with_decoding(
                                             provider_model_credentials.get(variable) or "",
                                             decoding_context,
                                         )
-                                    except ValueError:
-                                        pass
 
                             # cache provider model credentials
                             provider_model_credentials_cache.set(credentials=provider_model_credentials)

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -1,5 +1,6 @@
 """Paragraph index processor."""
 
+import contextlib
 import logging
 import re
 import uuid
@@ -459,11 +460,9 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
         # This is especially important for image-only chunks where text is empty or minimal
         if is_default_prompt:
             language_for_prompt = document_language or "the same language as the input content"
-            try:
+            # If default prompt doesn't have {language} placeholder, use it as-is
+            with contextlib.suppress(KeyError):
                 summary_prompt = summary_prompt.format(language=language_for_prompt)
-            except KeyError:
-                # If default prompt doesn't have {language} placeholder, use it as-is
-                pass
 
         model_manager = ModelManager.for_tenant(tenant_id=tenant_id)
         model_instance = model_manager.get_model_instance(

api/core/tools/workflow_as_tool/tool.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
 from collections.abc import Generator, Mapping, Sequence
@@ -170,10 +171,8 @@ class WorkflowTool(Tool):
 
         usage_metadata: dict[str, Any] = {}
         if total_tokens is not None:
-            try:
+            with contextlib.suppress(TypeError, ValueError):
                 usage_metadata["total_tokens"] = int(str(total_tokens))
-            except (TypeError, ValueError):
-                pass
         if total_price is not None:
             usage_metadata["total_price"] = str(total_price)
         currency = data.get("currency")

api/extensions/logstore/aliyun_logstore.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextlib
 import logging
 import os
 import socket
@@ -247,10 +248,9 @@ class AliyunLogStore:
         finally:
             # Ensure socket is properly closed
             if sock:
-                try:
+                # Ignore errors during cleanup
+                with contextlib.suppress(Exception):
                     sock.close()
-                except Exception:  # noqa: S110
-                    pass  # Ignore errors during cleanup
 
     @property
     def supports_pg_protocol(self) -> bool:

api/extensions/otel/celery_sqlcommenter.py

@@ -10,6 +10,7 @@ by opentelemetry.instrumentation.sqlcommenter_utils._add_framework_tags() when t
 SQLAlchemy instrumentor appends comments to SQL statements.
 """
 
+import contextlib
 import logging
 from typing import Any, TypedDict
 
@@ -105,10 +106,8 @@ def _on_task_postrun(*args: object, **kwargs: object) -> None:
     except Exception:
         logger.debug("Failed to detach SQL commenter context", exc_info=True)
     finally:
-        try:
+        with contextlib.suppress(AttributeError):
             delattr(task, _TOKEN_ATTR)
-        except AttributeError:
-            pass
 
 
 def setup_celery_sqlcommenter() -> None:

api/libs/broadcast_channel/redis/_subscription.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 import queue
 import threading
@@ -246,14 +247,10 @@ class RedisSubscriptionBase(Subscription):
         try:
             self._queue.put_nowait(SIG_CLOSE)
         except queue.Full:
-            try:
+            with contextlib.suppress(queue.Empty):
                 self._queue.get_nowait()
-            except queue.Empty:
-                pass
-            try:
+            with contextlib.suppress(queue.Full):
                 self._queue.put_nowait(SIG_CLOSE)
-            except queue.Full:
-                pass
 
     # Abstract methods to be implemented by subclasses
     def _get_subscription_type(self) -> str:

api/models/model.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextlib
 import json
 import re
 import uuid
@@ -1369,11 +1370,9 @@ class Conversation(Base):
             if not workflow_run:
                 continue
 
-            try:
+            # Handle invalid status values gracefully
+            with contextlib.suppress(ValueError, KeyError):
                 status_counts[WorkflowExecutionStatus(workflow_run.status)] += 1
-            except (ValueError, KeyError):
-                # Handle invalid status values gracefully
-                pass
 
         return {
             "success": status_counts[WorkflowExecutionStatus.SUCCEEDED],

api/providers/vdb/vdb-iris/src/dify_vdb_iris/iris_vector.py

@@ -10,7 +10,7 @@ import json
 import logging
 import threading
 import uuid
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from typing import TYPE_CHECKING, Any, override
 
 from sqlalchemy.orm import Session
@@ -104,10 +104,8 @@ class IrisConnectionPool:
             is_valid = True
         except (OSError, RuntimeError) as e:
             logger.debug("Connection validation failed: %s", e)
-            try:
+            with suppress(OSError, RuntimeError):
                 conn.close()
-            except (OSError, RuntimeError):
-                pass
 
         with self._lock:
             self._pool.append(conn if is_valid else self._create_connection())
@@ -168,10 +166,8 @@ class IrisConnectionPool:
         """Close all connections (application shutdown only)."""
         with self._lock:
             for conn in self._pool:
-                try:
+                with suppress(OSError, RuntimeError):
                     conn.close()
-                except (OSError, RuntimeError):
-                    pass
             self._pool.clear()
             self._in_use = 0
             self._schemas_initialized.clear()

api/services/knowledge_fs/background_engine.py

@@ -5,6 +5,7 @@ request at a time. Closing its stdin or killing the owner also kills the engine'
 database leases/checkpoints then govern recovery after Celery redelivery.
 """
 
+import contextlib
 import json
 import os
 import selectors
@@ -77,15 +78,11 @@ class KnowledgeFSBackgroundEngine:
             return
         # Never kill a process group inherited across a fork: it belongs to another Celery child.
         if self._owner_pid == os.getpid():
-            try:
+            with contextlib.suppress(ProcessLookupError):
                 os.killpg(process.pid, signal.SIGKILL)
-            except ProcessLookupError:
-                pass
-            try:
+            # Preserve the task's original exception; SIGKILL has already been requested.
+            with contextlib.suppress(subprocess.TimeoutExpired):
                 process.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                # Preserve the task's original exception; SIGKILL has already been requested.
-                pass
         for stream in (process.stdin, process.stdout):
             if stream is not None:
                 stream.close()

api/services/knowledge_fs/buffered_upload_admission.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from collections import deque
 from collections.abc import Generator
-from contextlib import AbstractContextManager, contextmanager
+from contextlib import AbstractContextManager, contextmanager, suppress
 from dataclasses import dataclass
 from threading import Condition
 from typing import Protocol
@@ -94,10 +94,8 @@ class KnowledgeFSBufferedUploadAdmission:
         )
 
     def _remove_waiter(self, waiter: object) -> None:
-        try:
+        with suppress(ValueError):
             self._waiters.remove(waiter)
-        except ValueError:
-            pass
 
 
 DEFAULT_KNOWLEDGE_FS_BUFFERED_UPLOAD_ADMISSION = KnowledgeFSBufferedUploadAdmission(

api/services/plugin/dependencies_analysis.py

@@ -1,3 +1,4 @@
+import contextlib
 import re
 from collections.abc import Mapping
 from typing import Any
@@ -70,10 +71,8 @@ class DependenciesAnalysisService:
                 provider = tool.get("provider_id") or tool.get("provider_name") or tool.get("provider")
                 if not isinstance(provider, str):
                     continue
-                try:
+                with contextlib.suppress(ValueError, NotFound):
                     dependencies.append(cls.analyze_tool_provider_reference(provider))
-                except (ValueError, NotFound):
-                    pass
         return dependencies
 
     @classmethod

api/services/retention/workflow_run/restore_archived_workflow_run.py

@@ -32,6 +32,8 @@ class ArchiveManifest(TypedDict, total=False):
 
 _manifest_adapter: TypeAdapter[ArchiveManifest] = TypeAdapter(ArchiveManifest)
 
+import contextlib
+
 from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
 
 from extensions.ext_database import db
@@ -338,10 +340,8 @@ class WorkflowRunRestore:
             if isinstance(column.type, DateTime):
                 value = result.get(column.key)
                 if isinstance(value, str):
-                    try:
+                    with contextlib.suppress(ValueError):
                         result[column.key] = datetime.fromisoformat(value)
-                    except ValueError:
-                        pass
 
         return result
 

api/services/skill_management_service.py

@@ -22,7 +22,7 @@ import xml.etree.ElementTree as ET
 import zipfile
 from base64 import b64encode
 from collections.abc import Generator
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from dataclasses import dataclass
 from datetime import datetime
 from enum import StrEnum
@@ -4708,10 +4708,8 @@ class SkillManagementService:
                 continue
 
             if source[index : index + 2] == "\\'":
-                try:
+                with suppress(ValueError):
                     text_parts.append(bytes.fromhex(source[index + 2 : index + 4]).decode("latin-1"))
-                except ValueError:
-                    pass
                 index += 4
                 continue
             if index + 1 < len(source):

api/services/workflow_event_snapshot_service.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import contextlib
 import json
 import logging
 import queue
@@ -661,10 +662,8 @@ def _start_buffering(subscription) -> BufferState:
                     buffer_state.queue.put_nowait(event)
                 except queue.Full:
                     dropped_count += 1
-                    try:
+                    with contextlib.suppress(queue.Empty):
                         buffer_state.queue.get_nowait()
-                    except queue.Empty:
-                        pass
                     try:
                         buffer_state.queue.put_nowait(event)
                     except queue.Full:

api/tests/unit_tests/core/mcp/session/test_base_session.py

@@ -1,3 +1,4 @@
+import contextlib
 import logging
 import queue
 import time
@@ -466,10 +467,8 @@ def test_check_receiver_status_fail(streams):
     future = executor.submit(raise_err)
     session._receiver_future = future
 
-    try:
+    with contextlib.suppress(BaseException):
         future.result()
-    except BaseException:
-        pass
 
     with pytest.raises(RuntimeError, match="Receiver failed"):
         session.check_receiver_status()

api/tests/unit_tests/core/tools/test_mcp_tool.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import base64
+import contextlib
 from unittest.mock import patch
 
 import pytest
@@ -288,8 +289,6 @@ def test_invoke_skips_forwarding_outside_enterprise_edition(config_overrides):
                 with patch("core.entities.mcp_provider.MCPProviderEntity.decrypt_server_url", return_value="u"):
                     with patch("core.entities.mcp_provider.MCPProviderEntity.decrypt_headers", return_value={}):
                         # Should not raise; should not call enterprise.
-                        try:
+                        with contextlib.suppress(Exception):
                             tool.invoke_remote_mcp_tool({}, user_id=None, app_id=None)
-                        except Exception:
-                            pass
         inject.assert_not_called()

api/tests/unit_tests/core/workflow/context/test_execution_context.py

@@ -2,7 +2,7 @@
 
 import contextvars
 import threading
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from typing import Any
 from unittest.mock import MagicMock
 
@@ -176,10 +176,8 @@ class TestExecutionContext:
                             return
             except Exception as exc:
                 errors.append(exc)
-                try:
+                with suppress(Exception):
                     barrier.abort()
-                except Exception:
-                    pass
 
         t1 = threading.Thread(target=worker)
         t2 = threading.Thread(target=worker)

api/tests/unit_tests/libs/test_schedule_utils_enhanced.py

@@ -5,6 +5,7 @@ These tests verify that the backend schedule_utils functions properly support
 the enhanced cron syntax introduced in the frontend, ensuring full compatibility.
 """
 
+import contextlib
 import unittest
 from datetime import UTC, datetime, timedelta
 
@@ -368,10 +369,8 @@ class TestPerformanceEnhanced(unittest.TestCase):
         # under xdist, where worker preemption can inflate elapsed time without
         # indicating a real regression in cron parsing.
         for expr in complex_expressions:
-            try:
+            with contextlib.suppress(Exception):
                 calculate_next_run_at(expr, "UTC", self.base_time)
-            except Exception:
-                pass
 
         start_time = time.process_time()