"""File operations for Microsoft Graph API (OneDrive/SharePoint)."""
from typing import Dict, List, Any
from .base import BaseOperation, OperationError
from ..graph_client import MicrosoftGraphClient
class FileOperations(BaseOperation):
"""Operations for file and document management."""
def get_supported_actions(self) -> List[str]:
"""Return supported file actions."""
return [
"list", "search", "get", "download", "upload",
"delete", "share", "get_permissions"
]
def _validate_action_params(self, action: str, params: Dict) -> None:
"""Validate parameters for file actions."""
# access_token is required for all actions (from TrustyVault)
self._require_param(params, "access_token", str)
self._require_param(params, "microsoft_user", str)
if action == "list":
# folder_path is optional
pass
elif action == "search":
query = self._require_param(params, "query", str)
if len(query) < 2:
raise OperationError(
code="INVALID_QUERY",
message="Search query must be at least 2 characters"
)
elif action in ["get", "download", "delete", "get_permissions"]:
self._require_param(params, "file_id", str)
elif action == "upload":
self._require_param(params, "file_name", str)
self._require_param(params, "content", str) # Base64 encoded
elif action == "share":
self._require_param(params, "file_id", str)
share_type = params.get("share_type", "view")
self._validate_choice(share_type, ["view", "edit"], "share_type")
def _execute_action(self, action: str, params: Dict) -> Any:
"""Execute file action."""
if action == "list":
return self._action_list(params)
elif action == "search":
return self._action_search(params)
elif action == "get":
return self._action_get(params)
elif action == "download":
return self._action_download(params)
elif action == "upload":
return self._action_upload(params)
elif action == "delete":
return self._action_delete(params)
elif action == "share":
return self._action_share(params)
elif action == "get_permissions":
return self._action_get_permissions(params)
def _action_list(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""List files in folder."""
folder_path = params.get("folder_path", "root")
max_results = params.get("max_results", 100)
try:
if folder_path == "root":
url = f"/users/{microsoft_user}/drive/root/children"
else:
url = f"/users/{microsoft_user}/drive/root:/{folder_path}:/children"
query_params = {
"$top": max_results,
"$select": "id,name,size,createdDateTime,lastModifiedDateTime,folder,file"
}
response = client.get(url, params=query_params)
items = response.get("value", [])
return {
"items": items,
"count": len(items),
"folder_path": folder_path
}
except Exception as e:
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list files: {str(e)}",
details={"folder_path": folder_path}
)
def _action_search(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Search files by query."""
query = params["query"]
max_results = params.get("max_results", 50)
try:
query_params = {
"$search": query,
"$top": max_results,
"$select": "id,name,size,createdDateTime,lastModifiedDateTime,webUrl"
}
response = client.get(f"/users/{microsoft_user}/drive/root/search(q='')", params=query_params)
items = response.get("value", [])
return {
"items": items,
"count": len(items),
"query": query
}
except Exception as e:
raise OperationError(
code="SEARCH_FAILED",
message=f"Failed to search files: {str(e)}",
details={"query": query}
)
def _action_get(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Get file metadata."""
file_id = params["file_id"]
try:
file_item = client.get(f"/users/{microsoft_user}/drive/items/{file_id}")
return {"file": file_item}
except Exception as e:
if "404" in str(e):
raise OperationError(
code="FILE_NOT_FOUND",
message=f"File not found: {file_id}",
details={"file_id": file_id}
)
raise OperationError(
code="GET_FAILED",
message=f"Failed to get file: {str(e)}",
details={"file_id": file_id}
)
def _action_download(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Get file download URL."""
file_id = params["file_id"]
try:
file_item = client.get(f"/users/{microsoft_user}/drive/items/{file_id}")
download_url = file_item.get("@microsoft.graph.downloadUrl")
if not download_url:
raise OperationError(
code="NO_DOWNLOAD_URL",
message="File has no download URL",
details={"file_id": file_id}
)
return {
"file_id": file_id,
"name": file_item.get("name"),
"download_url": download_url,
"size": file_item.get("size")
}
except OperationError:
raise
except Exception as e:
raise OperationError(
code="DOWNLOAD_FAILED",
message=f"Failed to get download URL: {str(e)}",
details={"file_id": file_id}
)
def _action_upload(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Upload file to OneDrive."""
file_name = params["file_name"]
content = params["content"] # Base64 or raw bytes
folder_path = params.get("folder_path", "root")
try:
if folder_path == "root":
url = f"/users/{microsoft_user}/drive/root:/{file_name}:/content"
else:
url = f"/users/{microsoft_user}/drive/root:/{folder_path}/{file_name}:/content"
# Upload content
uploaded_file = client.put(url, data=content)
return {
"uploaded": True,
"file_id": uploaded_file.get("id"),
"name": uploaded_file.get("name"),
"size": uploaded_file.get("size"),
"web_url": uploaded_file.get("webUrl")
}
except Exception as e:
raise OperationError(
code="UPLOAD_FAILED",
message=f"Failed to upload file: {str(e)}",
details={"file_name": file_name}
)
def _action_delete(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Delete file."""
file_id = params["file_id"]
try:
client.delete(f"/users/{microsoft_user}/drive/items/{file_id}")
return {
"deleted": True,
"file_id": file_id
}
except Exception as e:
raise OperationError(
code="DELETE_FAILED",
message=f"Failed to delete file: {str(e)}",
details={"file_id": file_id}
)
def _action_share(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Create sharing link for file."""
file_id = params["file_id"]
share_type = params.get("share_type", "view")
scope = params.get("scope", "anonymous") # anonymous or organization
try:
share_request = {
"type": share_type,
"scope": scope
}
permission = client.post(
f"/users/{microsoft_user}/drive/items/{file_id}/createLink",
json=share_request
)
return {
"file_id": file_id,
"share_link": permission.get("link", {}).get("webUrl"),
"type": share_type,
"scope": scope
}
except Exception as e:
raise OperationError(
code="SHARE_FAILED",
message=f"Failed to create sharing link: {str(e)}",
details={"file_id": file_id}
)
def _action_get_permissions(self, params: Dict) -> Dict:
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
client = MicrosoftGraphClient(access_token=access_token)
"""Get file permissions."""
file_id = params["file_id"]
try:
response = client.get(f"/users/{microsoft_user}/drive/items/{file_id}/permissions")
permissions = response.get("value", [])
return {
"file_id": file_id,
"permissions": permissions,
"count": len(permissions)
}
except Exception as e:
raise OperationError(
code="GET_PERMISSIONS_FAILED",
message=f"Failed to get permissions: {str(e)}",
details={"file_id": file_id}
)
