"""User operations for Microsoft Graph API."""
from typing import Dict, List, Any
from .base import BaseOperation, OperationError
from ..graph_client import MicrosoftGraphClient
class UserOperations(BaseOperation):
"""Operations for user management and search."""
def get_supported_actions(self) -> List[str]:
"""Return supported user actions."""
return ["search", "get", "list", "get_presence", "get_photo", "get_me"]
def _validate_action_params(self, action: str, params: Dict) -> None:
"""Validate parameters for user actions."""
# access_token is required for all actions (from TrustyVault)
self._require_param(params, "access_token", str)
if 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",
details={"query": query, "min_length": 2}
)
max_results = params.get("max_results", 10)
if not isinstance(max_results, int) or max_results < 1 or max_results > 100:
raise OperationError(
code="INVALID_PARAM",
message="max_results must be between 1 and 100",
details={"max_results": max_results}
)
elif action == "get":
# target_email or user_id optional (if omitted, uses microsoft_user)
if "target_email" in params:
self._validate_email(params["target_email"])
elif action == "list":
max_results = params.get("max_results", 50)
if not isinstance(max_results, int) or max_results < 1 or max_results > 999:
raise OperationError(
code="INVALID_PARAM",
message="max_results must be between 1 and 999",
details={"max_results": max_results}
)
elif action == "get_presence":
if "user_id" not in params and "email" not in params:
raise OperationError(
code="MISSING_PARAM",
message="Either 'user_id' or 'email' must be provided"
)
elif action == "get_photo":
if "user_id" not in params and "email" not in params:
raise OperationError(
code="MISSING_PARAM",
message="Either 'user_id' or 'email' must be provided"
)
def _execute_action(self, action: str, params: Dict) -> Any:
"""Execute user action."""
if action == "search":
return self._action_search(params)
elif action == "get":
return self._action_get(params)
elif action == "list":
return self._action_list(params)
elif action == "get_presence":
return self._action_get_presence(params)
elif action == "get_photo":
return self._action_get_photo(params)
elif action == "get_me":
return self._action_get_me(params)
def _action_search(self, params: Dict) -> Dict:
"""Search for users by name or email.
Uses Microsoft Graph $search parameter with ConsistencyLevel header.
Format: "displayName:value" OR "mail:value"
"""
query = params["query"]
access_token = params["access_token"]
max_results = params.get("max_results", 10)
# Build search query in proper format: "property:value"
# Try multiple fields with OR
search_query = f'"displayName:{query}" OR "givenName:{query}" OR "surname:{query}" OR "mail:{query}"'
# Execute search
url = "/users"
query_params = {
"$search": search_query,
"$select": "id,displayName,mail,jobTitle,department,officeLocation,userPrincipalName",
"$top": max_results,
"$orderby": "displayName"
}
headers = {
"ConsistencyLevel": "eventual"
}
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
response = client.get(url, params=query_params, headers=headers)
users = response.get("value", [])
return {
"users": users,
"count": len(users),
"query": query
}
except Exception as e:
raise OperationError(
code="SEARCH_FAILED",
message=f"User search failed: {str(e)}",
details={"query": query}
)
def _action_get(self, params: Dict) -> Dict:
"""Get user details by ID or email.
If target_email is omitted, returns the authenticated user's profile using /me endpoint.
If target_email is provided, searches for that user's profile.
Automatically resolves email display to UPN if needed.
"""
access_token = params["access_token"]
target_email = params.get("target_email")
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
# Determine URL and query params
query_params = {
"$select": "id,displayName,mail,jobTitle,department,officeLocation,"
"userPrincipalName,mobilePhone,businessPhones,givenName,surname"
}
if target_email:
# Search for specific user - resolve email to UPN if needed
identifier = self._resolve_user_identifier(target_email, client)
url = f"/users/{identifier}"
else:
# Get authenticated user's profile using /me endpoint
url = "/me"
try:
user = client.get(url, params=query_params)
return {"user": user}
except Exception as e:
error_msg = str(e)
if "ResourceNotFound" in error_msg or "404" in error_msg:
raise OperationError(
code="USER_NOT_FOUND",
message=f"User not found: {target_email or 'authenticated user'}",
details={"target_email": target_email, "url": url}
)
raise OperationError(
code="GET_FAILED",
message=f"Failed to get user: {error_msg}",
details={"target_email": target_email, "url": url}
)
def _resolve_user_identifier(self, email_or_upn: str, client: MicrosoftGraphClient) -> str:
"""Resolve email display to UPN or user ID.
Tries multiple strategies:
1. Direct lookup (works if already UPN or user ID)
2. Filter search by mail or userPrincipalName
3. Return original if all fails (will error later with clearer message)
"""
# Strategy 1: Try direct lookup
try:
user = client.get(f"/users/{email_or_upn}")
return user.get("userPrincipalName", email_or_upn)
except:
pass # Continue to next strategy
# Strategy 2: Search by mail or UPN using filter
try:
response = client.get(
f"/users?$filter=mail eq '{email_or_upn}' or userPrincipalName eq '{email_or_upn}'"
)
users = response.get("value", [])
if users:
return users[0].get("userPrincipalName", email_or_upn)
except:
pass # Continue to next strategy
# Strategy 3: Return original (will fail later with clear error)
return email_or_upn
def _action_list(self, params: Dict) -> Dict:
"""List users with optional filters."""
access_token = params["access_token"]
max_results = params.get("max_results", 50)
filters = params.get("filters", {})
# Build query
query_params = {
"$select": "id,displayName,mail,jobTitle,department,officeLocation,userPrincipalName",
"$top": max_results,
"$orderby": "displayName"
}
# Add filters
filter_parts = []
if "department" in filters:
filter_parts.append(f"department eq '{filters['department']}'")
if "job_title_contains" in filters:
filter_parts.append(f"startswith(jobTitle, '{filters['job_title_contains']}')")
if filter_parts:
query_params["$filter"] = " and ".join(filter_parts)
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
response = client.get("/users", params=query_params)
users = response.get("value", [])
return {
"users": users,
"count": len(users),
"filters": filters
}
except Exception as e:
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list users: {str(e)}",
details={"filters": filters}
)
def _action_get_presence(self, params: Dict) -> Dict:
"""Get user presence status (Available, Busy, etc.)."""
access_token = params["access_token"]
user_id = params.get("user_id")
email = params.get("email")
# First get user ID if email provided
if email and not user_id:
user_data = self._action_get({"email": email, "access_token": access_token})
user_id = user_data["user"]["id"]
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
presence = client.get(f"/users/{user_id}/presence")
return {
"user_id": user_id,
"availability": presence.get("availability"),
"activity": presence.get("activity")
}
except Exception as e:
raise OperationError(
code="PRESENCE_FAILED",
message=f"Failed to get presence: {str(e)}",
details={"user_id": user_id}
)
def _action_get_photo(self, params: Dict) -> Dict:
"""Get user profile photo metadata."""
access_token = params["access_token"]
user_id = params.get("user_id")
email = params.get("email")
identifier = user_id if user_id else email
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
# Get photo metadata
photo_meta = client.get(f"/users/{identifier}/photo")
return {
"user_id": identifier,
"photo": {
"width": photo_meta.get("width"),
"height": photo_meta.get("height"),
"id": photo_meta.get("id")
},
"photo_url": f"https://graph.microsoft.com/v1.0/users/{identifier}/photo/$value"
}
except Exception as e:
# Photo not found is common, return null
if "404" in str(e) or "ImageNotFound" in str(e):
return {
"user_id": identifier,
"photo": None,
"message": "User has no profile photo"
}
raise OperationError(
code="PHOTO_FAILED",
message=f"Failed to get photo: {str(e)}",
details={"user_id": identifier}
)
def _action_get_me(self, params: Dict) -> Dict:
"""Get authenticated user information.
This method uses the /me endpoint to retrieve information about
the user who owns the OAuth token being used for authentication.
This is THE definitive way to determine who the organizer is:
the user who authenticated via OAuth IS the organizer.
Returns:
Dict with user information:
- id: User ID (GUID)
- displayName: Full name
- mail: Email address
- userPrincipalName: UPN
- jobTitle: Job title
- department: Department
- officeLocation: Office location
- mobilePhone: Mobile phone
- businessPhones: Business phones
"""
access_token = params["access_token"]
url = "/me"
query_params = {
"$select": "id,displayName,mail,jobTitle,department,officeLocation,"
"userPrincipalName,mobilePhone,businessPhones,givenName,surname"
}
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
user = client.get(url, params=query_params)
return {
"user": user,
"is_authenticated": True,
"message": "This is the authenticated user (token owner)"
}
except Exception as e:
raise OperationError(
code="GET_ME_FAILED",
message=f"Failed to get authenticated user: {str(e)}",
details={"endpoint": "/me"}
)
