148 lines
5.6 KiB
Python
148 lines
5.6 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
|
|
|
|
@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)}"
|
|
)
|
|
|