Replace Exception with specific error types (#5755)

Replaces generic Exception raises with more specific error types such as RuntimeError, ValueError, NotImplementedError, ImportError, and custom exceptions across multiple modules. This improves error handling clarity and aligns with Python best practices for exception usage.

Author: FeodorFitsner

sdk/python/packages/flet-cli/src/flet_cli/utils/android_sdk.py

@@ -4,9 +4,10 @@
 from pathlib import Path
 from typing import Optional
 
+from rich.progress import Progress
+
 from flet_cli.utils import processes
 from flet_cli.utils.distros import download_with_progress, extract_with_progress
-from rich.progress import Progress
 
 ANDROID_CMDLINE_TOOLS_DOWNLOAD_VERSION = "11076708"
 ANDROID_CMDLINE_TOOLS_VERSION = "12.0"
@@ -98,7 +99,7 @@ def cmdline_tools_url(self):
                 },
             }[platform.system()][platform.machine()]
         except KeyError:
-            raise Exception(
+            raise RuntimeError(
                 f"Unsupported platform: {platform.system()}-{platform.machine()}"
             ) from None
 
@@ -183,7 +184,7 @@ def _install_package(self, home_dir: Path, package_name: str) -> int:
         )
         if p.returncode != 0:
             self.log(p.stderr)
-            raise Exception("Error installing Android SDK tools")
+            raise RuntimeError("Error installing Android SDK tools")
         return 1
 
     def _accept_licenses(self, home_dir: Path):
@@ -208,7 +209,7 @@ def _accept_licenses(self, home_dir: Path):
         )
         if p.returncode != 0:
             self.log(p.stderr)
-            raise Exception("Error accepting Android SDK licenses")
+            raise RuntimeError("Error accepting Android SDK licenses")
 
     def get_installed_packages(self, home_dir: Path):
         self.log("Checking installed Android APIs and build tools")
@@ -219,7 +220,7 @@ def get_installed_packages(self, home_dir: Path):
         )
         if p.returncode != 0:
             self.log(p.stderr)
-            raise Exception(
+            raise RuntimeError(
                 "Error retrieving the list of installed Android SDK packages"
             )
         return p.stdout

sdk/python/packages/flet-cli/src/flet_cli/utils/project_dependencies.py

@@ -8,7 +8,6 @@
 def get_poetry_dependencies(
     poetry_dependencies: Optional[dict[str, Any]] = None,
 ) -> Optional[list[str]]:
-
     if poetry_dependencies is None:
         return None
 
@@ -51,8 +50,9 @@ def format_dependency_version(dependency_name: str, dependency_value: Any):
                             dependency_name = ""
                             sep = ""
                         else:
-                            raise Exception(
-                                f"Unsupported dependency specification: {dependency_name} = {dependency_value}"
+                            raise ValueError(
+                                "Unsupported dependency specification: "
+                                f"{dependency_name} = {dependency_value}"
                             )
 
             # markers - common for all
@@ -88,7 +88,6 @@ def format_dependency_version(dependency_name: str, dependency_value: Any):
 def get_project_dependencies(
     project_dependencies: Optional[dict[str, Any]] = None,
 ) -> Optional[list[str]]:
-
     if project_dependencies is None:
         return None
 

sdk/python/packages/flet-web/src/flet_web/fastapi/flet_app.py

@@ -325,7 +325,7 @@ async def __on_message(self, data: Any):
 
         else:
             # it's something else
-            raise Exception(f'Unknown message "{action}": {body}')
+            raise RuntimeError(f'Unknown message "{action}": {body}')
 
         if task:
             self.__running_tasks.add(task)

sdk/python/packages/flet-web/src/flet_web/fastapi/flet_app_manager.py

@@ -6,11 +6,10 @@
 from datetime import datetime, timezone
 from typing import Optional
 
+import flet_web.fastapi as flet_fastapi
 from flet.messaging.connection import Connection
 from flet.messaging.session import Session
 from flet.pubsub.pubsub_hub import PubSubHub
-
-import flet_web.fastapi as flet_fastapi
 from flet_web.fastapi.oauth_state import OAuthState
 
 logger = logging.getLogger(flet_fastapi.__name__)
@@ -84,7 +83,7 @@ async def reconnect_session(self, session_id: str, conn: Connection):
             session = self.__sessions[session_id]
             await session.connect(conn)
         else:
-            raise Exception(f"Session has expired or not found: {session_id}")
+            raise RuntimeError(f"Session has expired or not found: {session_id}")
 
     async def disconnect_session(self, session_id: str, session_timeout_seconds: int):
         logger.info(f"Session disconnected: {session_id}")

sdk/python/packages/flet-web/src/flet_web/fastapi/flet_static_files.py

@@ -6,11 +6,11 @@
 from typing import Optional
 
 from fastapi.staticfiles import StaticFiles
-from flet.controls.types import RouteUrlStrategy, WebRenderer
-from flet.utils import Once, get_bool_env_var
 from starlette.types import Receive, Scope, Send
 
 import flet_web.fastapi as flet_fastapi
+from flet.controls.types import RouteUrlStrategy, WebRenderer
+from flet.utils import Once, get_bool_env_var
 from flet_web import (
     get_package_web_dir,
     patch_font_manifest_json,
@@ -120,7 +120,7 @@ async def __config(self, root_path: str):
         logger.info(f"Web root: {web_dir}")
 
         if not os.path.exists(web_dir):
-            raise Exception(f"Web root path not found: {web_dir}")
+            raise RuntimeError(f"Web root path not found: {web_dir}")
 
         # user-defined assets
         if self.__assets_dir:

sdk/python/packages/flet-web/src/flet_web/fastapi/flet_upload.py

@@ -3,9 +3,10 @@
 from datetime import datetime, timezone
 from typing import Optional
 
-import flet_web.fastapi as flet_fastapi
 from anyio import open_file
-from fastapi import Request
+from fastapi import HTTPException, Request, status
+
+import flet_web.fastapi as flet_fastapi
 from flet_web.uploads import build_upload_query_string, get_upload_signature
 
 logger = logging.getLogger(flet_fastapi.__name__)
@@ -18,7 +19,8 @@ class FletUpload:
     Parameters:
 
     * `upload_dir` (str) - an absolute path to a directory with uploaded files.
-    * `max_upload_size` (str, int) - maximum size of a single upload, bytes. Unlimited if `None`.
+    * `max_upload_size` (str, int) - maximum size of a single upload, bytes.
+       Unlimited if `None`.
     * `secret_key` (str, optional) - secret key to sign and verify upload requests.
     """
 
@@ -50,14 +52,24 @@ def __init__(
     """
 
     async def handle(self, request: Request):
-        file_name = request.query_params["f"]
-        expire_str = request.query_params["e"]
-        signature = request.query_params["s"]
+        query_params = request.query_params
+        file_name = query_params.get("f")
+        expire_str = query_params.get("e")
+        signature = query_params.get("s")
 
         if not file_name or not expire_str or not signature:
-            raise Exception("Invalid request")
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Missing upload parameters",
+            )
 
-        expire_date = datetime.fromisoformat(expire_str)
+        try:
+            expire_date = datetime.fromisoformat(expire_str)
+        except ValueError as e:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid expiration parameter",
+            ) from e
 
         # verify signature
         query_string = build_upload_query_string(file_name, expire_date)
@@ -67,17 +79,26 @@ async def handle(self, request: Request):
             )
             != signature
         ):
-            raise Exception("Invalid request")
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Invalid upload signature",
+            )
 
         # check expiration date
         if datetime.now(timezone.utc) >= expire_date:
-            raise Exception("Invalid request")
+            raise HTTPException(
+                status_code=status.HTTP_410_GONE,
+                detail="Upload URL has expired",
+            )
 
         # build/validate dest path
         joined_path = os.path.join(self.__upload_dir, file_name)
         full_path = os.path.realpath(joined_path)
         if os.path.commonpath([full_path, self.__upload_dir]) != self.__upload_dir:
-            raise Exception("Invalid request")
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Invalid upload destination",
+            )
 
         # create directory if not exists
         dest_dir = os.path.dirname(full_path)
@@ -89,7 +110,8 @@ async def handle(self, request: Request):
             async for chunk in request.stream():
                 size += len(chunk)
                 if self.__max_upload_size and size > self.__max_upload_size:
-                    raise Exception(
-                        f"Max upload size reached: {self.__max_upload_size}"
+                    raise HTTPException(
+                        status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+                        detail="Max upload size exceeded",
                     )
                 await f.write(chunk)

sdk/python/packages/flet-web/src/flet_web/uploads.py

@@ -39,8 +39,9 @@ def get_upload_signature(
     if env_secret_key:
         secret_key = env_secret_key
     if not secret_key:
-        raise Exception(
-            "Specify secret_key parameter or set FLET_SECRET_KEY environment variable to enable uploads."
+        raise RuntimeError(
+            "Specify secret_key parameter or set FLET_SECRET_KEY environment "
+            "variable to enable uploads."
         )
     signing_key = hmac.new(
         secret_key.encode("utf-8"),
@@ -49,6 +50,6 @@ def get_upload_signature(
     ).digest()
     return hmac.new(
         signing_key,
-        f"{upload_endpoint_path.strip('/')}{query_string}".encode("utf-8"),
+        f"{upload_endpoint_path.strip('/')}{query_string}".encode(),
         hashlib.sha256,
     ).hexdigest()

sdk/python/packages/flet/docs/cookbook/authentication.md

@@ -438,7 +438,7 @@ def main(page: Page):
 
     def on_login(e):
         if e.error:
-            raise Exception(e.error)
+            raise RuntimeError(e.error)
         print("User ID:", page.auth.user.id)
         print("Access token:", page.auth.token.access_token)
 

sdk/python/packages/flet/src/flet/auth/authorization_service.py

@@ -90,7 +90,7 @@ async def __fetch_user_and_groups(self):
             self.user = await self.provider._fetch_user(self.__token.access_token)
             if self.user is None and self.provider.user_endpoint is not None:
                 if self.provider.user_id_fn is None:
-                    raise Exception(
+                    raise ValueError(
                         "user_id_fn must be specified too if user_endpoint is not None"
                     )
                 self.user = await self.__get_user()

sdk/python/packages/flet/src/flet/auth/oauth_provider.py

@@ -36,7 +36,7 @@ def __init__(
         self.code_verifier = code_verifier
 
     def _name(self):
-        raise Exception("Not implemented")
+        raise NotImplementedError("Subclasses must implement _name()")
 
     async def _fetch_groups(self, access_token: str) -> list[Group]:
         return []