Merge pull request #716 from christianlouis/code-health/simplify-save-onedrive-settings-12338634662690804221

🧹 [Code Health] Simplify save_onedrive_settings logic and extract env_utils

app/api/onedrive.py

@@ -3,7 +3,6 @@ OneDrive API endpoints
 """
 
 import logging
-import os
 from datetime import datetime, timedelta
 from typing import Annotated, Optional
 
@@ -14,6 +13,7 @@ from sqlalchemy.orm import Session
 from app.auth import require_login
 from app.config import settings
 from app.database import get_db
+from app.utils.env_utils import update_env_file
 from app.utils.oauth_helper import exchange_oauth_token
 from app.utils.settings_service import save_setting_to_db
 from app.utils.settings_sync import notify_settings_updated
@@ -115,32 +115,7 @@ async def test_onedrive_token(request: Request):
             settings.onedrive_refresh_token = new_refresh_token
 
             # Also try to update .env file if it exists
-            try:
+            update_env_file({"ONEDRIVE_REFRESH_TOKEN": new_refresh_token})
-                env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
-                if os.path.exists(env_path):
-                    with open(env_path, "r") as f:
-                        env_lines = f.readlines()
-
-                    updated_lines = []
-                    updated = False
-
-                    for line in env_lines:
-                        if line.startswith("ONEDRIVE_REFRESH_TOKEN="):
-                            updated_lines.append(f"ONEDRIVE_REFRESH_TOKEN={new_refresh_token}\n")
-                            updated = True
-                        else:
-                            updated_lines.append(line)
-
-                    if not updated:
-                        updated_lines.append(f"ONEDRIVE_REFRESH_TOKEN={new_refresh_token}\n")
-
-                    with open(env_path, "w") as f:
-                        f.writelines(updated_lines)
-
-                    logger.info("Updated refresh token in .env file")
-
-            except Exception as e:
-                logger.warning(f"Failed to update refresh token in .env file: {e}")
 
             # Persist the rotated refresh token to the database
             try:
@@ -246,75 +221,26 @@ async def save_onedrive_settings(
             user.get("preferred_username") or user.get("username") or user.get("email") or user.get("id") or "wizard"
         )
 
-        # Best-effort .env file write
+        # Build settings dictionary mapped to database/memory keys
-        try:
+        onedrive_settings = {
-            env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
+            "onedrive_refresh_token": refresh_token,
-            if not os.path.exists(env_path):
+            "onedrive_client_id": client_id,
-                logger.warning(f".env file not found at {env_path}, skipping file write")
+            "onedrive_client_secret": client_secret,
-            else:
+            "onedrive_tenant_id": tenant_id,
-                logger.info(f"Updating OneDrive settings in {env_path}")
+            "onedrive_folder_path": folder_path,
+        }
 
-                with open(env_path, "r") as f:
+        # Filter out None values
-                    env_lines = f.readlines()
+        onedrive_settings = {k: v for k, v in onedrive_settings.items() if v is not None}
 
-                onedrive_settings = {"ONEDRIVE_REFRESH_TOKEN": refresh_token}
+        # Best-effort .env file write using the new utility
-                if client_id:
+        env_settings = {k.upper(): v for k, v in onedrive_settings.items()}
-                    onedrive_settings["ONEDRIVE_CLIENT_ID"] = client_id
+        update_env_file(env_settings)
-                if client_secret:
-                    onedrive_settings["ONEDRIVE_CLIENT_SECRET"] = client_secret
-                if tenant_id:
-                    onedrive_settings["ONEDRIVE_TENANT_ID"] = tenant_id
-                if folder_path:
-                    onedrive_settings["ONEDRIVE_FOLDER_PATH"] = folder_path
 
-                updated = set()
+        # Update in-memory settings and persist to database dynamically
-                new_env_lines = []
+        for key, value in onedrive_settings.items():
-                for line in env_lines:
+            setattr(settings, key, value)
-                    stripped_line = line.rstrip()
+            save_setting_to_db(db, key, value, changed_by=changed_by)
-                    is_updated = False
-                    for key, value in onedrive_settings.items():
-                        if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="):
-                            new_env_lines.append(f"{key}={value}")
-                            updated.add(key)
-                            is_updated = True
-                            break
-                    if not is_updated:
-                        new_env_lines.append(stripped_line)
-
-                for key, value in onedrive_settings.items():
-                    if key not in updated:
-                        new_env_lines.append(f"{key}={value}")
-
-                with open(env_path, "w") as f:
-                    f.write("\n".join(new_env_lines) + "\n")
-
-                logger.info("Successfully updated OneDrive settings in .env file")
-        except Exception as env_err:
-            logger.warning(f"Failed to write .env file (non-fatal): {env_err}")
-
-        # Update the settings in memory
-        if refresh_token:
-            settings.onedrive_refresh_token = refresh_token
-        if client_id:
-            settings.onedrive_client_id = client_id
-        if client_secret:
-            settings.onedrive_client_secret = client_secret
-        if tenant_id:
-            settings.onedrive_tenant_id = tenant_id
-        if folder_path:
-            settings.onedrive_folder_path = folder_path
-
-        # Persist to database (primary)
-        if refresh_token:
-            save_setting_to_db(db, "onedrive_refresh_token", refresh_token, changed_by=changed_by)
-        if client_id:
-            save_setting_to_db(db, "onedrive_client_id", client_id, changed_by=changed_by)
-        if client_secret:
-            save_setting_to_db(db, "onedrive_client_secret", client_secret, changed_by=changed_by)
-        if tenant_id:
-            save_setting_to_db(db, "onedrive_tenant_id", tenant_id, changed_by=changed_by)
-        if folder_path:
-            save_setting_to_db(db, "onedrive_folder_path", folder_path, changed_by=changed_by)
 
         notify_settings_updated()
 

app/utils/env_utils.py

@@ -0,0 +1,55 @@
+import logging
+import os
+from typing import Dict
+
+logger = logging.getLogger(__name__)
+
+
+def update_env_file(settings_to_update: Dict[str, str]) -> bool:
+    """
+    Updates the .env file with the given settings (best-effort).
+    Creates or modifies existing keys.
+
+    Args:
+        settings_to_update: A dictionary mapping uppercase env var names to their new string values.
+
+    Returns:
+        True if the file was successfully updated, False otherwise.
+    """
+    try:
+        env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
+        if not os.path.exists(env_path):
+            logger.warning(f".env file not found at {env_path}, skipping file write")
+            return False
+
+        logger.info(f"Updating settings in {env_path}")
+
+        with open(env_path, "r") as f:
+            env_lines = f.readlines()
+
+        updated = set()
+        new_env_lines = []
+        for line in env_lines:
+            stripped_line = line.rstrip()
+            is_updated = False
+            for key, value in settings_to_update.items():
+                if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="):
+                    new_env_lines.append(f"{key}={value}")
+                    updated.add(key)
+                    is_updated = True
+                    break
+            if not is_updated:
+                new_env_lines.append(stripped_line)
+
+        for key, value in settings_to_update.items():
+            if key not in updated:
+                new_env_lines.append(f"{key}={value}")
+
+        with open(env_path, "w") as f:
+            f.write("\n".join(new_env_lines) + "\n")
+
+        logger.info("Successfully updated settings in .env file")
+        return True
+    except Exception as env_err:
+        logger.warning(f"Failed to write .env file (non-fatal): {env_err}")
+        return False

tests/test_api_onedrive_coverage.py

@@ -75,8 +75,8 @@ class TestTestTokenRotation:
             patch.object(settings, "onedrive_refresh_token", "old_token"),
             patch.object(settings, "onedrive_client_id", "cid"),
             patch.object(settings, "onedrive_client_secret", "sec"),
-            patch("app.api.onedrive.os.path.join", return_value=str(env_file)),
+            patch("app.utils.env_utils.os.path.join", return_value=str(env_file)),
-            patch("app.api.onedrive.os.path.exists", return_value=True),
+            patch("app.utils.env_utils.os.path.exists", return_value=True),
             patch("app.database.SessionLocal") as mock_session_local,
             patch("app.api.onedrive.save_setting_to_db"),
             patch("app.api.onedrive.notify_settings_updated"),
@@ -117,7 +117,7 @@ class TestTestTokenRotation:
             patch.object(settings, "onedrive_refresh_token", "old_token"),
             patch.object(settings, "onedrive_client_id", "cid"),
             patch.object(settings, "onedrive_client_secret", "sec"),
-            patch("app.api.onedrive.os.path.exists", return_value=False),
+            patch("app.utils.env_utils.os.path.exists", return_value=False),
             patch("app.database.SessionLocal") as mock_session_local,
             patch("app.api.onedrive.save_setting_to_db"),
             patch("app.api.onedrive.notify_settings_updated"),
@@ -157,7 +157,7 @@ class TestTestTokenRotation:
             patch.object(settings, "onedrive_refresh_token", "old_token"),
             patch.object(settings, "onedrive_client_id", "cid"),
             patch.object(settings, "onedrive_client_secret", "sec"),
-            patch("app.api.onedrive.os.path.exists", return_value=True),
+            patch("app.utils.env_utils.os.path.exists", return_value=True),
             patch("builtins.open", side_effect=PermissionError("Permission denied")),
             patch("app.database.SessionLocal") as mock_session_local,
             patch("app.api.onedrive.save_setting_to_db"),
@@ -198,7 +198,7 @@ class TestTestTokenRotation:
             patch.object(settings, "onedrive_refresh_token", "old_token"),
             patch.object(settings, "onedrive_client_id", "cid"),
             patch.object(settings, "onedrive_client_secret", "sec"),
-            patch("app.api.onedrive.os.path.exists", return_value=False),
+            patch("app.utils.env_utils.os.path.exists", return_value=False),
             patch("app.database.SessionLocal", side_effect=Exception("DB error")),
         ):
             response = client.get("/api/onedrive/test-token")
@@ -277,8 +277,8 @@ class TestTokenRotationEnvAppendLine:
             patch.object(settings, "onedrive_refresh_token", "old_token"),
             patch.object(settings, "onedrive_client_id", "cid"),
             patch.object(settings, "onedrive_client_secret", "sec"),
-            patch("app.api.onedrive.os.path.join", return_value=str(env_file)),
+            patch("app.utils.env_utils.os.path.join", return_value=str(env_file)),
-            patch("app.api.onedrive.os.path.exists", return_value=True),
+            patch("app.utils.env_utils.os.path.exists", return_value=True),
             patch("app.database.SessionLocal") as mock_sl,
             patch("app.api.onedrive.save_setting_to_db"),
             patch("app.api.onedrive.notify_settings_updated"),