refactor: clean up structural dispatch patterns

examples/github_trigger/events/code_scanning_alert/code_scanning_alert.py

@@ -91,7 +91,6 @@ def _check_branch(self, alert: Mapping[str, Any], value: str | None) -> None:
             or alert.get("ref")
             or ""
         )
-        # Convert refs/heads/main -> main
-        branch = ref.split("/", 2)[-1] if ref.startswith("refs/heads/") else ref
+        branch = ref.removeprefix("refs/heads/")
         if branch not in branches:
             raise EventIgnoreError

examples/github_trigger/events/push/push.py

@@ -62,11 +62,7 @@ def _check_branch(
             return
 
         current_ref = payload.get("ref") or ""
-        branch = (
-            current_ref.split("/", 2)[-1]
-            if current_ref.startswith("refs/heads/")
-            else current_ref
-        )
+        branch = current_ref.removeprefix("refs/heads/")
         if branch not in allowed_branches:
             raise EventIgnoreError
 

examples/github_trigger/events/secret_scanning/secret_scanning.py

@@ -87,6 +87,6 @@ def _check_branch(self, payload: Mapping[str, Any], value: str | None) -> None:
         ref = (payload.get("alert") or {}).get("most_recent_instance", {}).get(
             "ref"
         ) or ""
-        branch = ref.split("/", 2)[-1] if ref.startswith("refs/heads/") else ref
+        branch = ref.removeprefix("refs/heads/")
         if branch and branch not in branches:
             raise EventIgnoreError

examples/google_calendar_trigger/provider/google_calendar_trigger.py

@@ -35,6 +35,7 @@
 FALSY_STRINGS = frozenset({"false", "0", "no", "off"})
 WATCH_SUCCESS_STATUSES = frozenset({HTTPStatus.OK, HTTPStatus.CREATED})
 CHANNEL_STOP_SUCCESS_STATUSES = frozenset({HTTPStatus.OK, HTTPStatus.NO_CONTENT})
+RFC3339_UTC_SUFFIX = "Z"
 
 
 class SyncTokenExpiredError(TriggerError):
@@ -146,8 +147,8 @@ def _parse_rfc3339(value: object) -> datetime.datetime | None:
         return None
     try:
         text = value.strip()
-        if text.endswith("Z"):
-            text = text[:-1] + "+00:00"
+        if text.endswith(RFC3339_UTC_SUFFIX):
+            text = f"{text.removesuffix(RFC3339_UTC_SUFFIX)}+00:00"
         return datetime.datetime.fromisoformat(text)
     except Exception:
         return None

examples/notion_datasource/datasources/utils/notion_client.py

@@ -474,6 +474,35 @@ def format_rich_text(self, content: str) -> list[dict[str, Any]]:
         """
         return self.create_rich_text(content)
 
+    def _normalize_icon(
+        self, page_icon: dict[str, Any] | None
+    ) -> dict[str, str] | None:
+        if not page_icon:
+            return None
+
+        icon_type = page_icon["type"]
+        if icon_type in FILE_ICON_TYPES:
+            url = page_icon[icon_type]["url"]
+            return {
+                "type": "url",
+                "url": url if url.startswith("http") else f"https://www.notion.so{url}",
+            }
+
+        return {"type": "emoji", "emoji": page_icon[icon_type]}
+
+    def _resolve_parent_id(self, access_token: str, parent: dict[str, Any]) -> str:
+        parent_type = parent["type"]
+        match parent_type:
+            case "block_id":
+                return self._resolve_block_parent_page_id(
+                    access_token,
+                    parent[parent_type],
+                )
+            case "workspace":
+                return "root"
+            case _:
+                return parent[parent_type]
+
     def get_authorized_pages(self) -> list[OnlineDocumentPage]:
         pages: list[OnlineDocumentPage] = []
         access_token = self.integration_token
@@ -489,33 +518,10 @@ def get_authorized_pages(self) -> list[OnlineDocumentPage]:
                     and page_result["properties"][key]["title"]
                 ):
                     title_list = page_result["properties"][key]["title"]
-                    if len(title_list) > 0 and "plain_text" in title_list[0]:
+                    if title_list and "plain_text" in title_list[0]:
                         page_name = title_list[0]["plain_text"]
-            page_icon = page_result["icon"]
-            if page_icon:
-                icon_type = page_icon["type"]
-                if icon_type in FILE_ICON_TYPES:
-                    url = page_icon[icon_type]["url"]
-                    icon = {
-                        "type": "url",
-                        "url": url
-                        if url.startswith("http")
-                        else f"https://www.notion.so{url}",
-                    }
-                else:
-                    icon = {"type": "emoji", "emoji": page_icon[icon_type]}
-            else:
-                icon = None
-            parent = page_result["parent"]
-            parent_type = parent["type"]
-            if parent_type == "block_id":
-                parent_id = self._resolve_block_parent_page_id(
-                    access_token, parent[parent_type]
-                )
-            elif parent_type == "workspace":
-                parent_id = "root"
-            else:
-                parent_id = parent[parent_type]
+            icon = self._normalize_icon(page_result["icon"])
+            parent_id = self._resolve_parent_id(access_token, page_result["parent"])
             page = OnlineDocumentPage(
                 page_id=page_id,
                 page_name=page_name,
@@ -530,35 +536,12 @@ def get_authorized_pages(self) -> list[OnlineDocumentPage]:
             page_id = database_result["id"]
             page_name = (
                 database_result["title"][0]["plain_text"]
-                if len(database_result["title"]) > 0
+                if database_result["title"]
                 else "Untitled"
             )
 
-            page_icon = database_result["icon"]
-            if page_icon:
-                icon_type = page_icon["type"]
-                if icon_type in FILE_ICON_TYPES:
-                    url = page_icon[icon_type]["url"]
-                    icon = {
-                        "type": "url",
-                        "url": url
-                        if url.startswith("http")
-                        else f"https://www.notion.so{url}",
-                    }
-                else:
-                    icon = {"type": "emoji", "emoji": page_icon[icon_type]}
-            else:
-                icon = None
-            parent = database_result["parent"]
-            parent_type = parent["type"]
-            if parent_type == "block_id":
-                parent_id = self._resolve_block_parent_page_id(
-                    access_token, parent[parent_type]
-                )
-            elif parent_type == "workspace":
-                parent_id = "root"
-            else:
-                parent_id = parent[parent_type]
+            icon = self._normalize_icon(database_result["icon"])
+            parent_id = self._resolve_parent_id(access_token, database_result["parent"])
             page = OnlineDocumentPage(
                 page_id=page_id,
                 page_name=page_name,

examples/notion_datasource/datasources/utils/notion_extractor.py

@@ -304,40 +304,35 @@ def _format_page_data(self, page_data: dict[str, Any]) -> dict[str, Any]:
             prop_type = prop_data.get("type")
 
             # Extract value based on property type
-            if prop_type == "title":
-                title_content = prop_data.get("title", [])
-                value = self._client.extract_plain_text(title_content)
-                if value:
-                    title = value  # Save title for the result
-            elif prop_type == "rich_text":
-                text_content = prop_data.get("rich_text", [])
-                value = self._client.extract_plain_text(text_content)
-            elif prop_type == "number":
-                value = prop_data.get("number")
-            elif prop_type == "select":
-                select_data = prop_data.get("select", {})
-                value = select_data.get("name") if select_data else None
-            elif prop_type == "multi_select":
-                multi_select = prop_data.get("multi_select", [])
-                value = (
-                    [item.get("name") for item in multi_select] if multi_select else []
-                )
-            elif prop_type == "date":
-                date_data = prop_data.get("date", {})
-                start = date_data.get("start") if date_data else None
-                end = date_data.get("end") if date_data else None
-                value = {"start": start, "end": end} if start else None
-            elif prop_type == "checkbox":
-                value = prop_data.get("checkbox")
-            elif prop_type == "url":
-                value = prop_data.get("url")
-            elif prop_type == "email":
-                value = prop_data.get("email")
-            elif prop_type == "phone_number":
-                value = prop_data.get("phone_number")
-            else:
-                # For other property types, just note the type
-                value = f"<{prop_type}>"
+            match prop_type:
+                case "title":
+                    title_content = prop_data.get("title", [])
+                    value = self._client.extract_plain_text(title_content)
+                    if value:
+                        title = value  # Save title for the result
+                case "rich_text":
+                    text_content = prop_data.get("rich_text", [])
+                    value = self._client.extract_plain_text(text_content)
+                case "number" | "checkbox" | "url" | "email" | "phone_number":
+                    value = prop_data.get(prop_type)
+                case "select":
+                    select_data = prop_data.get("select", {})
+                    value = select_data.get("name") if select_data else None
+                case "multi_select":
+                    multi_select = prop_data.get("multi_select", [])
+                    value = (
+                        [item.get("name") for item in multi_select]
+                        if multi_select
+                        else []
+                    )
+                case "date":
+                    date_data = prop_data.get("date", {})
+                    start = date_data.get("start") if date_data else None
+                    end = date_data.get("end") if date_data else None
+                    value = {"start": start, "end": end} if start else None
+                case _:
+                    # For other property types, just note the type
+                    value = f"<{prop_type}>"
 
             formatted_properties[prop_name] = value
 

src/dify_plugin/core/documentation/generator.py

@@ -262,12 +262,8 @@ def _extract_referenced_types(self, field_type: object) -> set[type]:
         if isinstance(field_type, type):
             if issubclass(field_type, (BaseModel, Enum)):
                 referenced.add(field_type)
-        # Handle generic types (List, Dict, Union, etc)
-        elif (
-            hasattr(field_type, "__origin__") and field_type.__origin__ == Union
-        ) or hasattr(field_type, "__args__"):
-            # Handle Union types
-            for arg in field_type.__args__:
+        else:
+            for arg in get_args(field_type):
                 referenced.update(self._extract_referenced_types(arg))
 
         return referenced
@@ -347,22 +343,13 @@ def _is_container_type(
         container_types: tuple[type, ...] = (list, set),
     ) -> bool:
         """Check if a field type is a container type (list, set, etc)."""
-        try:
-            return (
-                hasattr(field_type, "__origin__")
-                and isinstance(getattr(field_type, "__origin__", None), type)
-                and getattr(field_type, "__origin__", None) in container_types
-            )
-        except Exception:
-            return False
+        origin = get_origin(field_type)
+        return isinstance(origin, type) and origin in container_types
 
     def _get_container_name(self, field_type: object) -> str:
         """Get the name of a container type."""
-        try:
-            origin = getattr(field_type, "__origin__", None)
-            return origin.__name__ if origin else str(field_type)
-        except Exception:
-            return str(field_type)
+        origin = get_origin(field_type)
+        return origin.__name__ if isinstance(origin, type) else str(field_type)
 
     def _write_schema_doc(self, f: TextIO, type_: type) -> None:
         """Write documentation for a single schema."""

src/dify_plugin/core/entities/invocation.py

@@ -30,12 +30,5 @@ def value_of(cls, value: str) -> "InvokeType":
 
         Returns:
             The return value.
-
-        Raises:
-            ValueError: If input values are invalid.
         """
-        for mode in cls:
-            if mode.value == value:
-                return mode
-        msg = f"invalid type value {value}"
-        raise ValueError(msg)
+        return cls(value)

src/dify_plugin/core/entities/plugin/io.py

@@ -13,11 +13,7 @@ class PluginInStreamEvent(Enum):
 
     @classmethod
     def value_of(cls, v: str) -> "PluginInStreamEvent":
-        for e in cls:
-            if e.value == v:
-                return e
-        msg = f"Invalid value for PluginInStream.Event: {v}"
-        raise ValueError(msg)
+        return cls(v)
 
 
 @dataclass(frozen=True, slots=True)

src/dify_plugin/core/entities/plugin/request.py

@@ -11,13 +11,9 @@
 )
 from dify_plugin.entities.model import EmbeddingInputType, ModelType
 from dify_plugin.entities.model.message import (
-    AssistantPromptMessage,
     PromptMessage,
-    PromptMessageRole,
     PromptMessageTool,
-    SystemPromptMessage,
-    ToolPromptMessage,
-    UserPromptMessage,
+    ensure_prompt_message,
 )
 from dify_plugin.entities.model.text_embedding import MultiModalContent
 from dify_plugin.entities.provider_config import CredentialType
@@ -171,22 +167,7 @@ def convert_prompt_messages(cls, v: list[object]) -> list[PromptMessage]:
             msg = "prompt_messages must be a list"
             raise TypeError(msg)
 
-        for i in range(len(v)):
-            if isinstance(v[i], PromptMessage):
-                continue
-
-            if v[i]["role"] == PromptMessageRole.USER.value:
-                v[i] = UserPromptMessage(**v[i])
-            elif v[i]["role"] == PromptMessageRole.ASSISTANT.value:
-                v[i] = AssistantPromptMessage(**v[i])
-            elif v[i]["role"] == PromptMessageRole.SYSTEM.value:
-                v[i] = SystemPromptMessage(**v[i])
-            elif v[i]["role"] == PromptMessageRole.TOOL.value:
-                v[i] = ToolPromptMessage(**v[i])
-            else:
-                v[i] = PromptMessage(**v[i])
-
-        return v
+        return [ensure_prompt_message(item) for item in v]
 
 
 class ModelInvokeLLMRequest(PluginAccessModelRequest, PromptMessageMixin):

src/dify_plugin/entities/model/llm.py

@@ -34,15 +34,8 @@ def value_of(cls, value: str) -> "LLMMode":
 
         Returns:
             The return value.
-
-        Raises:
-            ValueError: If input values are invalid.
         """
-        for mode in cls:
-            if mode.value == value:
-                return mode
-        msg = f"invalid mode value {value}"
-        raise ValueError(msg)
+        return cls(value)
 
 
 class LLMPollingStatus(StrEnum):