gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #815 from christianlouis/chore/refactor-save-settings-3579323758629064412

Browse Source 
🧹 refactor: simplify save settings endpoints and fix naming
This commit is contained in:
Christian Krakau-Louis
2026-03-23 18:59:47 +01:00
committed by GitHub
parent ea2dad0c08 52e8e535ff
commit fc1365dfec
9 changed files with 193 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/dropbox.py
+22 -39
View File
@@ -14,7 +14,7 @@ from app.auth import require_login
from app.config import settings
from app.database import get_db
from app.utils.oauth_helper import exchange_oauth_token
from app.utils.settings_service import save_setting_to_db
from app.utils.settings_service import save_setting_to_db, update_env_file
from app.utils.settings_sync import notify_settings_updated
# Set up logging
@@ -23,6 +23,17 @@ logger = logging.getLogger(__name__)
router = APIRouter()
def _require_admin(request: Request) -> dict:
"""Dependency to ensure the current user is an admin."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
@router.post("/dropbox/exchange-token")
@require_login
async def exchange_dropbox_token(
@@ -211,9 +222,9 @@ async def test_dropbox_token(request: Request):
@router.post("/dropbox/save-settings")
@require_login
async def save_dropbox_settings(
request: Request,
_admin: AdminUser,
refresh_token: Annotated[str, Form(...)],
app_key: Annotated[Optional[str], Form()] = None,
app_secret: Annotated[Optional[str], Form()] = None,
@@ -252,44 +263,16 @@ async def save_dropbox_settings(
# Best-effort .env file write
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")
else:
logger.info(f"Updating Dropbox settings in {env_path}")
dropbox_settings = {"DROPBOX_REFRESH_TOKEN": refresh_token}
if app_key:
dropbox_settings["DROPBOX_APP_KEY"] = app_key
if app_secret:
dropbox_settings["DROPBOX_APP_SECRET"] = app_secret
if folder_path:
dropbox_settings["DROPBOX_FOLDER"] = folder_path
with open(env_path, "r") as f:
env_lines = f.readlines()
dropbox_settings = {"DROPBOX_REFRESH_TOKEN": refresh_token}
if app_key:
dropbox_settings["DROPBOX_APP_KEY"] = app_key
if app_secret:
dropbox_settings["DROPBOX_APP_SECRET"] = app_secret
if folder_path:
dropbox_settings["DROPBOX_FOLDER"] = folder_path
updated = set()
new_env_lines = []
for line in env_lines:
stripped_line = line.rstrip()
is_updated = False
for key, value in dropbox_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 dropbox_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 Dropbox settings in .env file")
if not update_env_file(env_path, dropbox_settings):
logger.info("Continuing with in-memory update despite .env file update failure or skip")
except Exception as env_err:
logger.warning(f"Failed to write .env file (non-fatal): {env_err}")
app/api/google_drive.py
+17 -43
View File
@@ -14,7 +14,7 @@ from app.auth import require_login
from app.config import settings
from app.database import get_db
from app.utils.oauth_helper import exchange_oauth_token
from app.utils.settings_service import save_setting_to_db
from app.utils.settings_service import save_setting_to_db, update_env_file
from app.utils.settings_sync import notify_settings_updated
# Set up logging
@@ -23,6 +23,17 @@ logger = logging.getLogger(__name__)
router = APIRouter()
def _require_admin(request: Request) -> dict:
"""Dependency to ensure the current user is an admin."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
@router.post("/google-drive/exchange-token")
@require_login
async def exchange_google_drive_token(
@@ -362,9 +373,9 @@ def format_time_remaining(time_delta):
@router.post("/google-drive/save-settings")
@require_login
async def save_google_drive_settings(
request: Request,
_admin: AdminUser,
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
client_secret: Annotated[Optional[str], Form()] = None,
@@ -404,46 +415,9 @@ async def save_google_drive_settings(
drive_settings["GOOGLE_DRIVE_FOLDER_ID"] = folder_id
# Try to update the .env file, but don't fail if it doesn't exist (for Docker containers)
if os.path.exists(env_path):
try:
logger.info(f"Updating Google Drive settings in {env_path}")
# Read the current .env file
with open(env_path, "r") as f:
env_lines = f.readlines()
# Process each line and update or add settings
updated = set()
new_env_lines = []
for line in env_lines:
stripped_line = line.rstrip()
is_updated = False
for key, value in drive_settings.items():
if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="):
# Uncomment if commented out - check the original stripped line
new_env_lines.append(f"{key}={value}")
updated.add(key)
is_updated = True
break
if not is_updated:
new_env_lines.append(stripped_line)
# Add any settings that weren't updated (they weren't in the file)
for key, value in drive_settings.items():
if key not in updated:
new_env_lines.append(f"{key}={value}")
# Write the updated .env file
with open(env_path, "w") as f:
f.write("\n".join(new_env_lines) + "\n")
logger.info("Successfully updated Google Drive settings in .env file")
except Exception as e:
logger.warning(f"Failed to update .env file: {str(e)}, but will continue with in-memory update")
else:
logger.warning(
f".env file not found at {env_path}, skipping file update but continuing with in-memory update"
)
env_write_success = update_env_file(env_path, drive_settings)
if not env_write_success:
logger.info("Continuing with in-memory update despite .env file update failure or skip")
# Update the settings in memory (this always happens)
if refresh_token:
@@ -481,7 +455,7 @@ async def save_google_drive_settings(
return {
"status": "success",
"message": "Google Drive settings have been saved",
"in_memory_only": not os.path.exists(env_path),
"in_memory_only": not env_write_success,
}
except Exception as e:
app/api/onedrive.py
+49 -20
View File
@@ -3,6 +3,7 @@ OneDrive API endpoints
"""
import logging
import os
from datetime import datetime, timedelta
from typing import Annotated, Optional
@@ -13,9 +14,8 @@ 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_service import save_setting_to_db, update_env_file
from app.utils.settings_sync import notify_settings_updated
# Set up logging
@@ -24,6 +24,17 @@ logger = logging.getLogger(__name__)
router = APIRouter()
def _require_admin(request: Request) -> dict:
"""Dependency to ensure the current user is an admin."""
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
return user
AdminUser = Annotated[dict, Depends(_require_admin)]
@router.post("/onedrive/exchange-token")
@require_login
async def exchange_onedrive_token(
@@ -204,9 +215,9 @@ def format_time_remaining(time_delta):
@router.post("/onedrive/save-settings")
@require_login
async def save_onedrive_settings(
request: Request,
_admin: AdminUser,
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
client_secret: Annotated[Optional[str], Form()] = None,
@@ -223,26 +234,44 @@ async def save_onedrive_settings(
user.get("preferred_username") or user.get("username") or user.get("email") or user.get("id") or "wizard"
)
# Build settings dictionary mapped to database/memory keys
onedrive_settings = {
"onedrive_refresh_token": refresh_token,
"onedrive_client_id": client_id,
"onedrive_client_secret": client_secret,
"onedrive_tenant_id": tenant_id,
"onedrive_folder_path": folder_path,
}
# Best-effort .env file write
env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
onedrive_settings = {"ONEDRIVE_REFRESH_TOKEN": refresh_token}
if client_id:
onedrive_settings["ONEDRIVE_CLIENT_ID"] = client_id
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
# Filter out None values
onedrive_settings = {k: v for k, v in onedrive_settings.items() if v is not None}
if not update_env_file(env_path, onedrive_settings):
logger.info("Continuing with in-memory update despite .env file update failure or skip")
# Best-effort .env file write using the new utility
env_settings = {k.upper(): v for k, v in onedrive_settings.items()}
update_env_file(env_settings)
# 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
# Update in-memory settings and persist to database dynamically
for key, value in onedrive_settings.items():
setattr(settings, key, value)
save_setting_to_db(db, key, value, changed_by=changed_by)
# 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
+1 -2
View File
@@ -1,11 +1,10 @@
import logging
import os
from typing import Dict
logger = logging.getLogger(__name__)
def update_env_file(settings_to_update: Dict[str, str]) -> bool:
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.
app/utils/settings_service.py
+58
View File
@@ -8,6 +8,7 @@ This module provides functionality to:
"""
import logging
import os
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy.exc import SQLAlchemyError
@@ -3371,3 +3372,60 @@ def get_settings_for_export(db: Session, source: str = "db") -> Dict[str, str]:
# DB only
db_settings = get_all_settings_from_db(db)
return {k.upper(): v for k, v in sorted(db_settings.items()) if v is not None}
def update_env_file(env_path: str, settings_to_update: dict[str, str]) -> bool:
"""
Update an .env file with new settings.
Reads the file, updates matching settings (even if commented),
appends any that weren't found, and writes the result back.
Args:
env_path: Path to the .env file
settings_to_update: Dictionary mapping setting names (e.g. 'GOOGLE_DRIVE_USE_OAUTH') to string values
Returns:
True if the file was successfully updated, False otherwise (e.g. file not found or write error)
"""
if not os.path.exists(env_path):
logger.warning(f".env file not found at {env_path}, skipping file update")
return False
try:
logger.info(f"Updating settings in {env_path}")
# Read the current .env file
with open(env_path, "r") as f:
env_lines = f.readlines()
# Process each line and update or add settings
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}="):
# Uncomment if commented out - check the original stripped line
new_env_lines.append(f"{key}={value}")
updated.add(key)
is_updated = True
break
if not is_updated:
new_env_lines.append(stripped_line)
# Add any settings that weren't updated (they weren't in the file)
for key, value in settings_to_update.items():
if key not in updated:
new_env_lines.append(f"{key}={value}")
# Write the updated .env file
with open(env_path, "w") as f:
f.write("\n".join(new_env_lines) + "\n")
logger.info(f"Successfully updated settings in {env_path}")
return True
except Exception as e:
logger.warning(f"Failed to update {env_path}: {str(e)}")
return False
tests/test_api_google_drive_comprehensive.py
+9
View File
@@ -360,6 +360,15 @@ class TestFormatTimeRemaining:
class TestSaveGoogleDriveSettings:
"""Tests for POST /google-drive/save-settings endpoint."""
@pytest.fixture(autouse=True)
def _admin_override(self):
from app.api.google_drive import _require_admin
from app.main import app as fastapi_app
fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
yield
fastapi_app.dependency_overrides.pop(_require_admin, None)
@patch("builtins.open", new_callable=mock_open, read_data="# Existing config\n")
@patch("os.path.exists")
@patch("os.path.dirname")
tests/test_api_google_drive_coverage.py
+9
View File
@@ -195,6 +195,15 @@ class TestGetGoogleDriveTokenInfo:
class TestSaveGoogleDriveSettings:
"""Test save_google_drive_settings endpoint edge cases."""
@pytest.fixture(autouse=True)
def _admin_override(self):
from app.api.google_drive import _require_admin
from app.main import app as fastapi_app
fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
yield
fastapi_app.dependency_overrides.pop(_require_admin, None)
@patch("app.api.google_drive.settings")
@patch("os.path.exists")
def test_save_settings_env_file_not_exists(self, mock_exists, mock_settings, client: TestClient):
tests/test_api_google_drive_final.py
+10 -4
View File
@@ -152,11 +152,16 @@ class TestGetTokenInfoCredentialsBranches:
@pytest.mark.unit
class TestSaveGoogleDriveSettingsFalsyFields:
"""Cover branches 395->397, 449->451, 468->470 in save_google_drive_settings.
"""Cover branches 395->397, 449->451, 468->470 in save_google_drive_settings."""
Note: the Google Drive save endpoint is named save_google_drive_settings in the
source (app/api/google_drive.py).
"""
@pytest.fixture(autouse=True)
def _admin_override(self):
from app.api.google_drive import _require_admin
from app.main import app as fastapi_app
fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
yield
fastapi_app.dependency_overrides.pop(_require_admin, None)
@patch("app.api.google_drive.settings")
@patch("os.path.exists", return_value=False)
@@ -177,6 +182,7 @@ class TestSaveGoogleDriveSettingsFalsyFields:
with patch("app.api.google_drive.notify_settings_updated"):
result = await save_google_drive_settings(
request=mock_request,
_admin={"is_admin": True},
refresh_token="", # falsy → branches 395->397 and 449->451
client_id="cid",
client_secret=None,
tests/test_wizard_db_persist.py
+18
View File
@@ -129,6 +129,15 @@ class TestSetupWizardUndoSkip:
class TestDropboxSaveSettingsDbPersist:
"""Unit tests for save_dropbox_settings DB persistence."""
@pytest.fixture(autouse=True)
def _admin_override(self):
from app.api.dropbox import _require_admin
from app.main import app as fastapi_app
fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
yield
fastapi_app.dependency_overrides.pop(_require_admin, None)
@patch("app.api.dropbox.settings")
@patch("app.api.dropbox.notify_settings_updated")
@patch("app.api.dropbox.save_setting_to_db")
@@ -268,6 +277,15 @@ class TestGoogleDriveUpdateSettingsDbPersist:
class TestOneDriveSaveSettingsDbPersist:
"""Unit tests for save_onedrive_settings DB persistence."""
@pytest.fixture(autouse=True)
def _admin_override(self):
from app.api.onedrive import _require_admin
from app.main import app as fastapi_app
fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True}
yield
fastapi_app.dependency_overrides.pop(_require_admin, None)
@patch("app.api.onedrive.settings")
@patch("app.api.onedrive.notify_settings_updated")
@patch("app.api.onedrive.save_setting_to_db")