398 lines
16 KiB
Python
398 lines
16 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
SharePoint routes for folder browsing
|
|
Provides endpoints for listing SharePoint sites and browsing folders
|
|
"""
|
|
|
|
import logging
|
|
from typing import List, Dict, Any, Optional
|
|
from fastapi import APIRouter, HTTPException, Depends, Path, Query, Request, status
|
|
|
|
from modules.auth import limiter, getCurrentUser
|
|
from modules.datamodels.datamodelUam import User, UserConnection
|
|
from modules.interfaces.interfaceDbApp import getInterface
|
|
from modules.services import getInterface as getServices
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(
|
|
prefix="/api/sharepoint",
|
|
tags=["SharePoint"],
|
|
responses={
|
|
404: {"description": "Not found"},
|
|
400: {"description": "Bad request"},
|
|
401: {"description": "Unauthorized"},
|
|
500: {"description": "Internal server error"}
|
|
}
|
|
)
|
|
|
|
def _getUserConnection(interface, connectionId: str, userId: str) -> Optional[UserConnection]:
|
|
"""Get a user connection by ID, ensuring it belongs to the user"""
|
|
try:
|
|
connections = interface.getUserConnections(userId)
|
|
for conn in connections:
|
|
if conn.id == connectionId:
|
|
return conn
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user connection: {str(e)}")
|
|
return None
|
|
|
|
def _getUserConnectionByReference(interface, connectionReference: str, userId: str) -> Optional[UserConnection]:
|
|
"""
|
|
Get a user connection by reference string (format: connection:authority:username).
|
|
|
|
Args:
|
|
interface: Database interface
|
|
connectionReference: Reference string like 'connection:msft:user@email.com'
|
|
userId: User ID to verify ownership
|
|
|
|
Returns:
|
|
UserConnection if found and belongs to user, None otherwise
|
|
"""
|
|
try:
|
|
# Parse reference format: connection:{authority}:{username} [status:..., token:...]
|
|
# Remove state information if present
|
|
baseReference = connectionReference.split(' [')[0]
|
|
|
|
parts = baseReference.split(':')
|
|
if len(parts) < 3 or parts[0] != "connection":
|
|
logger.warning(f"Invalid connection reference format: {connectionReference}")
|
|
return None
|
|
|
|
authority = parts[1] # e.g., 'msft'
|
|
username = ':'.join(parts[2:]) # Handle usernames with colons
|
|
|
|
# Get user connections and find matching one
|
|
connections = interface.getUserConnections(userId)
|
|
for conn in connections:
|
|
connAuthority = conn.authority.value if hasattr(conn.authority, 'value') else str(conn.authority)
|
|
if connAuthority.lower() == authority.lower() and conn.externalUsername == username:
|
|
return conn
|
|
|
|
logger.debug(f"No connection found for reference: {connectionReference}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user connection by reference: {str(e)}")
|
|
return None
|
|
|
|
@router.get("/{connectionId}/sites", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("30/minute")
|
|
async def get_sharepoint_sites(
|
|
request: Request,
|
|
connectionId: str = Path(..., description="Microsoft connection ID"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> List[Dict[str, Any]]:
|
|
"""Get all SharePoint sites accessible via a Microsoft connection"""
|
|
try:
|
|
interface = getInterface(currentUser)
|
|
|
|
# Get the connection and verify it belongs to the user
|
|
connection = _getUserConnection(interface, connectionId, currentUser.id)
|
|
if not connection:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Connection {connectionId} not found or does not belong to user"
|
|
)
|
|
|
|
# Verify it's a Microsoft connection
|
|
authority = connection.authority.value if hasattr(connection.authority, 'value') else str(connection.authority)
|
|
if authority.lower() != 'msft':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Connection {connectionId} is not a Microsoft connection"
|
|
)
|
|
|
|
# Initialize services
|
|
services = getServices(currentUser, None)
|
|
|
|
# Set access token on SharePoint service
|
|
if not services.sharepoint.setAccessTokenFromConnection(connection):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Failed to set SharePoint access token. Connection may be expired or invalid."
|
|
)
|
|
|
|
# Discover SharePoint sites
|
|
sites = await services.sharepoint.discoverSites()
|
|
|
|
return sites
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting SharePoint sites: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error getting SharePoint sites: {str(e)}"
|
|
)
|
|
|
|
@router.get("/{connectionId}/sites/{siteId}/folders", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("60/minute")
|
|
async def list_sharepoint_folders(
|
|
request: Request,
|
|
connectionId: str = Path(..., description="Microsoft connection ID"),
|
|
siteId: str = Path(..., description="SharePoint site ID"),
|
|
path: Optional[str] = Query(None, description="Folder path (empty for root)"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> List[Dict[str, Any]]:
|
|
"""List folder contents for a SharePoint site and folder path"""
|
|
try:
|
|
interface = getInterface(currentUser)
|
|
|
|
# Get the connection and verify it belongs to the user
|
|
connection = _getUserConnection(interface, connectionId, currentUser.id)
|
|
if not connection:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Connection {connectionId} not found or does not belong to user"
|
|
)
|
|
|
|
# Verify it's a Microsoft connection
|
|
authority = connection.authority.value if hasattr(connection.authority, 'value') else str(connection.authority)
|
|
if authority.lower() != 'msft':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Connection {connectionId} is not a Microsoft connection"
|
|
)
|
|
|
|
# Initialize services
|
|
services = getServices(currentUser, None)
|
|
|
|
# Set access token on SharePoint service
|
|
if not services.sharepoint.setAccessTokenFromConnection(connection):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Failed to set SharePoint access token. Connection may be expired or invalid."
|
|
)
|
|
|
|
# Normalize folder path (empty string for root)
|
|
folderPath = path or ''
|
|
|
|
# List folder contents
|
|
items = await services.sharepoint.listFolderContents(siteId, folderPath)
|
|
|
|
return items or []
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error listing SharePoint folders: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error listing SharePoint folders: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/{connectionId}/folder-options", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("30/minute")
|
|
async def getSharepointFolderOptions(
|
|
request: Request,
|
|
connectionId: str = Path(..., description="Microsoft connection ID"),
|
|
siteId: Optional[str] = Query(None, description="Specific site ID to browse (if omitted, returns sites only)"),
|
|
path: Optional[str] = Query(None, description="Folder path within site to browse"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get SharePoint folders formatted as dropdown options.
|
|
|
|
Two modes:
|
|
1. If siteId is not provided: Returns list of sites (for site selection)
|
|
2. If siteId is provided: Returns folders within that site (optionally at specific path)
|
|
|
|
This avoids expensive iteration through all sites and folders.
|
|
"""
|
|
try:
|
|
interface = getInterface(currentUser)
|
|
|
|
# Get the connection and verify it belongs to the user
|
|
connection = _getUserConnection(interface, connectionId, currentUser.id)
|
|
if not connection:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Connection {connectionId} not found or does not belong to user"
|
|
)
|
|
|
|
# Verify it's a Microsoft connection
|
|
authority = connection.authority.value if hasattr(connection.authority, 'value') else str(connection.authority)
|
|
if authority.lower() != 'msft':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Connection {connectionId} is not a Microsoft connection"
|
|
)
|
|
|
|
# Initialize services
|
|
services = getServices(currentUser, None)
|
|
|
|
# Set access token on SharePoint service
|
|
if not services.sharepoint.setAccessTokenFromConnection(connection):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Failed to set SharePoint access token. Connection may be expired or invalid."
|
|
)
|
|
|
|
# Mode 1: Return sites list if no siteId specified
|
|
if not siteId:
|
|
sites = await services.sharepoint.discoverSites()
|
|
return [
|
|
{
|
|
"type": "site",
|
|
"value": site.get("id"),
|
|
"label": site.get("displayName", "Unknown Site"),
|
|
"siteId": site.get("id"),
|
|
"siteName": site.get("displayName", "Unknown Site"),
|
|
"webUrl": site.get("webUrl", ""),
|
|
"path": _extractSitePath(site.get("webUrl", ""))
|
|
}
|
|
for site in sites
|
|
]
|
|
|
|
# Mode 2: Return folders within specific site
|
|
folderPath = path or ""
|
|
items = await services.sharepoint.listFolderContents(siteId, folderPath)
|
|
|
|
if not items:
|
|
return []
|
|
|
|
folderOptions = []
|
|
for item in items:
|
|
if item.get("type") == "folder":
|
|
folderName = item.get("name", "")
|
|
itemPath = f"{folderPath}/{folderName}" if folderPath else folderName
|
|
|
|
folderOptions.append({
|
|
"type": "folder",
|
|
"value": itemPath,
|
|
"label": folderName,
|
|
"siteId": siteId,
|
|
"folderName": folderName,
|
|
"path": itemPath,
|
|
"hasChildren": True # Assume folders may have children
|
|
})
|
|
|
|
return folderOptions
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting SharePoint folder options: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error getting SharePoint folder options: {str(e)}"
|
|
)
|
|
|
|
|
|
def _extractSitePath(webUrl: str) -> str:
|
|
"""Extract site path from webUrl (e.g., https://company.sharepoint.com/sites/MySite -> /sites/MySite)"""
|
|
if "/sites/" in webUrl:
|
|
return "/sites/" + webUrl.split("/sites/")[1].split("/")[0]
|
|
return ""
|
|
|
|
|
|
# ============================================================================
|
|
# Universal folder-options endpoint (by connectionReference)
|
|
# ============================================================================
|
|
|
|
@router.get("/folder-options", response_model=List[Dict[str, Any]])
|
|
@limiter.limit("30/minute")
|
|
async def getSharepointFolderOptionsByReference(
|
|
request: Request,
|
|
connectionReference: str = Query(..., description="Connection reference string (e.g., 'connection:msft:user@email.com')"),
|
|
siteId: Optional[str] = Query(None, description="Specific site ID to browse (if omitted, returns sites only)"),
|
|
path: Optional[str] = Query(None, description="Folder path within site to browse"),
|
|
currentUser: User = Depends(getCurrentUser)
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get SharePoint folders formatted as dropdown options (universal endpoint).
|
|
|
|
Uses connectionReference instead of connectionId for easier integration.
|
|
|
|
Two modes:
|
|
1. If siteId is not provided: Returns list of sites (for site selection)
|
|
2. If siteId is provided: Returns folders within that site (optionally at specific path)
|
|
|
|
Args:
|
|
connectionReference: Connection reference string (e.g., 'connection:msft:user@email.com')
|
|
siteId: Optional site ID to browse folders within
|
|
path: Optional folder path within site
|
|
"""
|
|
try:
|
|
interface = getInterface(currentUser)
|
|
|
|
# Get the connection by reference
|
|
connection = _getUserConnectionByReference(interface, connectionReference, currentUser.id)
|
|
if not connection:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Connection not found for reference: {connectionReference}"
|
|
)
|
|
|
|
# Verify it's a Microsoft connection
|
|
authority = connection.authority.value if hasattr(connection.authority, 'value') else str(connection.authority)
|
|
if authority.lower() != 'msft':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Connection is not a Microsoft connection (authority: {authority})"
|
|
)
|
|
|
|
# Initialize services
|
|
services = getServices(currentUser, None)
|
|
|
|
# Set access token on SharePoint service
|
|
if not services.sharepoint.setAccessTokenFromConnection(connection):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Failed to set SharePoint access token. Connection may be expired or invalid."
|
|
)
|
|
|
|
# Mode 1: Return sites list if no siteId specified
|
|
if not siteId:
|
|
sites = await services.sharepoint.discoverSites()
|
|
return [
|
|
{
|
|
"type": "site",
|
|
"value": site.get("id"),
|
|
"label": site.get("displayName", "Unknown Site"),
|
|
"siteId": site.get("id"),
|
|
"siteName": site.get("displayName", "Unknown Site"),
|
|
"webUrl": site.get("webUrl", ""),
|
|
"path": _extractSitePath(site.get("webUrl", ""))
|
|
}
|
|
for site in sites
|
|
]
|
|
|
|
# Mode 2: Return folders within specific site
|
|
folderPath = path or ""
|
|
items = await services.sharepoint.listFolderContents(siteId, folderPath)
|
|
|
|
if not items:
|
|
return []
|
|
|
|
folderOptions = []
|
|
for item in items:
|
|
if item.get("type") == "folder":
|
|
folderName = item.get("name", "")
|
|
itemPath = f"{folderPath}/{folderName}" if folderPath else folderName
|
|
|
|
folderOptions.append({
|
|
"type": "folder",
|
|
"value": itemPath,
|
|
"label": folderName,
|
|
"siteId": siteId,
|
|
"folderName": folderName,
|
|
"path": itemPath,
|
|
"hasChildren": True # Assume folders may have children
|
|
})
|
|
|
|
return folderOptions
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting SharePoint folder options by reference: {str(e)}")
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error getting SharePoint folder options: {str(e)}"
|
|
)
|
|
|