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

style: fix linting issues and remove unused imports

Browse Source 
- Remove unused imports from all modified files
- Fix flake8 violations (unused variables, f-strings without placeholders)
- Apply Black formatting consistently
- Shorten long line in google_drive.py

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-08 08:29:38 +00:00
parent 551b23a80c
commit d2eb9846d3
6 changed files with 350 additions and 452 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/google_drive.py
+115 -138
View File
@@ -1,13 +1,12 @@
"""
Google Drive API endpoints
"""
from fastapi import APIRouter, Request, HTTPException, status, Form
import logging
import os
import requests
import json
from typing import Optional
from datetime import datetime, timedelta
from datetime import datetime
from app.auth import require_login
from app.config import settings
@@ -18,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/google-drive/exchange-token")
@require_login
async def exchange_google_drive_token(
@@ -26,7 +26,7 @@ async def exchange_google_drive_token(
client_secret: str = Form(...),
redirect_uri: str = Form(...),
code: str = Form(...),
folder_id: Optional[str] = Form(None)
folder_id: Optional[str] = Form(None),
):
"""
Exchange an authorization code for refresh and access tokens from Google.
@@ -34,29 +34,26 @@ async def exchange_google_drive_token(
"""
# Prepare the token request
token_url = "https://oauth2.googleapis.com/token"
payload = {
'client_id': client_id,
'client_secret': client_secret,
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
"client_id": client_id,
"client_secret": client_secret,
"code": code,
"redirect_uri": redirect_uri,
"grant_type": "authorization_code",
}
# Use shared OAuth helper (handles secure logging and error handling)
token_data = exchange_oauth_token(
provider_name="Google Drive",
token_url=token_url,
payload=payload
)
token_data = exchange_oauth_token(provider_name="Google Drive", token_url=token_url, payload=payload)
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"access_token": token_data["access_token"],
"expires_in": token_data.get("expires_in", 3600)
"expires_in": token_data.get("expires_in", 3600),
}
@router.post("/google-drive/update-settings")
@require_login
async def update_google_drive_settings(
@@ -65,50 +62,48 @@ async def update_google_drive_settings(
client_secret: str = Form(None),
refresh_token: str = Form(...),
folder_id: str = Form(None),
use_oauth: str = Form("true")
use_oauth: str = Form("true"),
):
"""
Update Google Drive settings in memory
"""
try:
logger.info("Updating Google Drive settings in memory")
# Convert use_oauth string to boolean
use_oauth_bool = use_oauth.lower() in ("true", "1", "yes", "y", "t")
# Update settings in memory
if refresh_token:
settings.google_drive_refresh_token = refresh_token
logger.info("Updated GOOGLE_DRIVE_REFRESH_TOKEN in memory")
if client_id:
settings.google_drive_client_id = client_id
logger.info("Updated GOOGLE_DRIVE_CLIENT_ID in memory")
if client_secret:
settings.google_drive_client_secret = client_secret
logger.info("Updated GOOGLE_DRIVE_CLIENT_SECRET in memory")
if folder_id:
settings.google_drive_folder_id = folder_id
logger.info("Updated GOOGLE_DRIVE_FOLDER_ID in memory")
# Set the OAuth flag
settings.google_drive_use_oauth = use_oauth_bool
logger.info(f"Updated GOOGLE_DRIVE_USE_OAUTH in memory to {use_oauth_bool}")
return {
"status": "success",
"message": "Google Drive settings have been updated in memory"
}
return {"status": "success", "message": "Google Drive settings have been updated in memory"}
except Exception as e:
logger.exception(f"Unexpected error updating Google Drive settings: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to update Google Drive settings: {str(e)}"
detail=f"Failed to update Google Drive settings: {str(e)}",
)
@router.get("/google-drive/test-token")
@require_login
async def test_google_drive_token(request: Request):
@@ -118,124 +113,112 @@ async def test_google_drive_token(request: Request):
"""
try:
from app.tasks.upload_to_google_drive import get_drive_service_oauth, get_google_drive_service
logger.info("Testing Google Drive token validity")
# Check if OAuth is enabled and configured
if getattr(settings, 'google_drive_use_oauth', False):
if not (settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token):
if getattr(settings, "google_drive_use_oauth", False):
if not (
settings.google_drive_client_id
and settings.google_drive_client_secret
and settings.google_drive_refresh_token
):
logger.warning("Google Drive OAuth credentials not fully configured")
return {
"status": "error",
"message": "Google Drive OAuth credentials are not fully configured"
}
return {"status": "error", "message": "Google Drive OAuth credentials are not fully configured"}
try:
# Test OAuth connection
service = get_drive_service_oauth()
# Get credentials for checking token validity
import google.oauth2.credentials
from google.auth.transport.requests import Request
credentials = google.oauth2.credentials.Credentials(
token=None,
refresh_token=settings.google_drive_refresh_token,
token_uri="https://oauth2.googleapis.com/token",
client_id=settings.google_drive_client_id,
client_secret=settings.google_drive_client_secret
client_secret=settings.google_drive_client_secret,
)
# Force a refresh to update the token expiration
if not credentials.valid:
credentials.refresh(Request())
# Get token expiration info
expiration_info = {}
if hasattr(credentials, 'expiry') and credentials.expiry:
if hasattr(credentials, "expiry") and credentials.expiry:
now = datetime.now()
expiry = credentials.expiry
time_left = expiry - now
expiration_info = {
"expires_at": expiry.isoformat(),
"expires_in_seconds": max(0, int(time_left.total_seconds())),
"expires_in_human": format_time_remaining(time_left)
"expires_in_human": format_time_remaining(time_left),
}
# Test basic API operation
about = service.about().get(fields="user").execute()
user_email = about.get("user", {}).get("emailAddress", "Unknown")
logger.info(f"Successfully connected to Google Drive as {user_email}")
return {
"status": "success",
"message": f"OAuth token is valid! Connected as {user_email}",
"account": user_email,
"auth_type": "oauth",
"token_info": expiration_info
"token_info": expiration_info,
}
except Exception as e:
error_msg = str(e)
logger.error(f"Google Drive OAuth token test failed: {error_msg}")
# Check if this is a token-related error
if "invalid_grant" in error_msg.lower() or "token" in error_msg.lower():
return {
"status": "error",
"message": f"OAuth token validation failed: {error_msg}",
"needs_reauth": True
"needs_reauth": True,
}
return {
"status": "error",
"message": f"Connection error: {error_msg}"
}
return {"status": "error", "message": f"Connection error: {error_msg}"}
else:
# Test service account connection
if not settings.google_drive_credentials_json:
logger.warning("Google Drive service account credentials not configured")
return {
"status": "error",
"message": "Google Drive service account credentials are not configured"
}
return {"status": "error", "message": "Google Drive service account credentials are not configured"}
try:
service = get_google_drive_service()
about = service.about().get(fields="user").execute()
# For service accounts, try to show the delegated user if available
user_email = about.get("user", {}).get("emailAddress", "Unknown")
delegated_user = getattr(settings, 'google_drive_delegate_to', None)
delegated_user = getattr(settings, "google_drive_delegate_to", None)
if delegated_user:
user_display = f"{user_email} (delegating as {delegated_user})"
else:
user_display = user_email
logger.info(f"Successfully connected to Google Drive using service account as {user_display}")
return {
"status": "success",
"message": f"Service account is valid! Connected as {user_display}",
"account": user_email,
"auth_type": "service_account"
"auth_type": "service_account",
}
except Exception as e:
error_msg = str(e)
logger.error(f"Google Drive service account test failed: {error_msg}")
return {
"status": "error",
"message": f"Service account validation failed: {error_msg}"
}
return {"status": "error", "message": f"Service account validation failed: {error_msg}"}
except Exception as e:
logger.exception("Unexpected error testing Google Drive token")
return {
"status": "error",
"message": f"Unexpected error: {str(e)}"
}
return {"status": "error", "message": f"Unexpected error: {str(e)}"}
@router.get("/google-drive/get-token-info")
@require_login
@@ -247,94 +230,88 @@ async def get_google_drive_token_info(request: Request):
"""
try:
logger.info("Getting Google Drive token information")
# Check if OAuth is enabled and configured
if not getattr(settings, 'google_drive_use_oauth', False):
if not getattr(settings, "google_drive_use_oauth", False):
logger.warning("OAuth is not enabled, using service account instead")
return {
"status": "error",
"message": "OAuth is not enabled. Service accounts don't support user-facing features like folder picker."
"message": "OAuth is not enabled. Service accounts don't support user-facing features.",
}
if not (settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token):
if not (
settings.google_drive_client_id
and settings.google_drive_client_secret
and settings.google_drive_refresh_token
):
logger.warning("Google Drive OAuth credentials not fully configured")
return {
"status": "error",
"message": "Google Drive OAuth credentials are not fully configured"
}
return {"status": "error", "message": "Google Drive OAuth credentials are not fully configured"}
try:
# Get credentials and access token
import google.oauth2.credentials
from google.auth.transport.requests import Request
credentials = google.oauth2.credentials.Credentials(
token=None,
refresh_token=settings.google_drive_refresh_token,
token_uri="https://oauth2.googleapis.com/token",
client_id=settings.google_drive_client_id,
client_secret=settings.google_drive_client_secret
client_secret=settings.google_drive_client_secret,
)
# Force a refresh to get a fresh access token
if not credentials.valid:
credentials.refresh(Request())
# Get token expiration info
expiration_info = {}
if hasattr(credentials, 'expiry') and credentials.expiry:
if hasattr(credentials, "expiry") and credentials.expiry:
now = datetime.now()
expiry = credentials.expiry
time_left = expiry - now
expiration_info = {
"expires_at": expiry.isoformat(),
"expires_in_seconds": max(0, int(time_left.total_seconds())),
"expires_in_human": format_time_remaining(time_left)
"expires_in_human": format_time_remaining(time_left),
}
# Return the token info
logger.info("Successfully retrieved Google Drive access token")
return {
"status": "success",
"message": "Access token successfully retrieved",
"access_token": credentials.token,
"token_info": expiration_info
"token_info": expiration_info,
}
except Exception as e:
error_msg = str(e)
logger.error(f"Failed to get Google Drive token: {error_msg}")
# Check if this is a token-related error
if "invalid_grant" in error_msg.lower() or "token" in error_msg.lower():
return {
"status": "error",
"message": f"OAuth token retrieval failed: {error_msg}",
"needs_reauth": True
"needs_reauth": True,
}
return {
"status": "error",
"message": f"Token retrieval error: {error_msg}"
}
return {"status": "error", "message": f"Token retrieval error: {error_msg}"}
except Exception as e:
logger.exception("Unexpected error getting Google Drive token info")
return {
"status": "error",
"message": f"Unexpected error: {str(e)}"
}
return {"status": "error", "message": f"Unexpected error: {str(e)}"}
def format_time_remaining(time_delta):
"""Format a timedelta into a human-readable string."""
if time_delta.total_seconds() <= 0:
return "Expired"
days = time_delta.days
hours, remainder = divmod(time_delta.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
parts = []
if days > 0:
parts.append(f"{days} day{'s' if days != 1 else ''}")
@@ -342,9 +319,10 @@ def format_time_remaining(time_delta):
parts.append(f"{hours} hour{'s' if hours != 1 else ''}")
if minutes > 0 and days == 0: # Only show minutes if less than a day
parts.append(f"{minutes} minute{'s' if minutes != 1 else ''}")
return ", ".join(parts)
@router.post("/google-drive/save-settings")
@require_login
async def save_dropbox_settings(
@@ -353,7 +331,7 @@ async def save_dropbox_settings(
client_secret: str = Form(None),
refresh_token: str = Form(...),
folder_id: str = Form(None),
use_oauth: str = Form("true")
use_oauth: str = Form("true"),
):
"""
Save Google Drive settings to the .env file
@@ -361,15 +339,13 @@ async def save_dropbox_settings(
try:
# Get the path to the .env file
env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
# Convert use_oauth string to boolean
use_oauth_bool = use_oauth.lower() in ("true", "1", "yes", "y", "t")
# Define settings to update
drive_settings = {
"GOOGLE_DRIVE_USE_OAUTH": str(use_oauth_bool).lower()
}
drive_settings = {"GOOGLE_DRIVE_USE_OAUTH": str(use_oauth_bool).lower()}
# Only update these if provided
if use_oauth_bool:
if refresh_token:
@@ -378,20 +354,20 @@ async def save_dropbox_settings(
drive_settings["GOOGLE_DRIVE_CLIENT_ID"] = client_id
if client_secret:
drive_settings["GOOGLE_DRIVE_CLIENT_SECRET"] = client_secret
# Always include folder ID if provided
if folder_id:
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 = []
@@ -408,22 +384,24 @@ async def save_dropbox_settings(
break
if not is_updated:
new_env_lines.append(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")
logger.warning(
f".env file not found at {env_path}, skipping file update but continuing with in-memory update"
)
# Update the settings in memory (this always happens)
if refresh_token:
settings.google_drive_refresh_token = refresh_token
@@ -433,21 +411,20 @@ async def save_dropbox_settings(
settings.google_drive_client_secret = client_secret
if folder_id:
settings.google_drive_folder_id = folder_id
# Set OAuth flag
settings.google_drive_use_oauth = use_oauth_bool
logger.info("Successfully updated Google Drive settings in memory")
return {
"status": "success",
"message": "Google Drive settings have been saved",
"in_memory_only": not os.path.exists(env_path)
"in_memory_only": not os.path.exists(env_path),
}
except Exception as e:
logger.exception(f"Unexpected error saving Google Drive settings: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to save Google Drive settings: {str(e)}"
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to save Google Drive settings: {str(e)}"
)