"""
Microsoft Bookings Operations - Standard API integration
Implements Microsoft Bookings API operations:
- Businesses management
- Services configuration
- Staff management
- Appointments CRUD
- Customers management
API Documentation: https://learn.microsoft.com/en-us/graph/api/resources/booking-api-overview
"""
from typing import Dict, List, Any
import logging
import asyncio
from .base import BaseOperation, OperationError
from .hybrid_booking_operations import HybridBookingOperations
logger = logging.getLogger(__name__)
def run_async(coro):
"""Run async coroutine in a sync context (event loop aware)"""
try:
loop = asyncio.get_running_loop()
# We're in an event loop, create a task and wait for it
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as pool:
return pool.submit(asyncio.run, coro).result()
except RuntimeError:
# No event loop running, use run_async()
return run_async(coro)
class BookingOperations(BaseOperation):
"""Microsoft Bookings operations using Graph API + Hybrid Booking workflow"""
area_name = "bookings"
def __init__(self, graph_client):
"""Initialize booking operations with Graph client"""
self.client = graph_client
self.hybrid_ops = HybridBookingOperations(graph_client)
def get_supported_actions(self) -> List[str]:
"""Return list of supported booking actions (MS API + Hybrid workflow)"""
return [
# Microsoft Bookings API
"list_businesses",
"get_business",
"list_services",
"get_service",
"list_staff",
"get_staff",
"list_appointments",
"get_appointment",
"create_appointment",
"update_appointment",
"cancel_appointment",
"list_customers",
"get_customer",
# Hybrid Booking Workflow
"create_hybrid_session",
"send_hybrid_invitation",
"get_hybrid_session",
"check_hybrid_slot",
"confirm_hybrid_booking",
"find_new_hybrid_slots"
]
def _validate_action_params(self, action: str, params: Dict) -> None:
"""Validate parameters for booking actions"""
if action == "list_businesses":
pass # No params required
elif action == "get_business":
if "business_id" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="business_id is required"
)
elif action in ["list_services", "list_staff", "list_appointments", "list_customers"]:
if "business_id" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="business_id is required"
)
elif action in ["get_service", "get_staff", "get_appointment", "get_customer"]:
if "business_id" not in params or "item_id" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="business_id and item_id are required"
)
elif action == "create_appointment":
required = ["business_id", "service_id", "customer_id", "start_datetime"]
for param in required:
if param not in params:
raise OperationError(
code="MISSING_PARAMETER",
message=f"{param} is required"
)
elif action in ["update_appointment", "cancel_appointment"]:
if "business_id" not in params or "appointment_id" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="business_id and appointment_id are required"
)
# Hybrid Booking Workflow validation
elif action == "create_hybrid_session":
required = ["organizer_email", "organizer_name", "external_email", "proposed_slots"]
for param in required:
if param not in params:
raise OperationError(
code="MISSING_PARAMETER",
message=f"{param} is required"
)
elif action in ["send_hybrid_invitation", "get_hybrid_session", "find_new_hybrid_slots"]:
if "session_id" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="session_id is required"
)
elif action == "check_hybrid_slot":
if "session_id" not in params or "slot_index" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="session_id and slot_index are required"
)
elif action == "confirm_hybrid_booking":
if "session_id" not in params or "selected_slot_index" not in params:
raise OperationError(
code="MISSING_PARAMETER",
message="session_id and selected_slot_index are required"
)
def _execute_action(self, action: str, params: Dict) -> Any:
"""Execute booking action (dispatcher)"""
if action == "list_businesses":
return run_async(self.list_businesses())
elif action == "get_business":
return run_async(self.get_business(params["business_id"]))
elif action == "list_services":
return run_async(self.list_services(params["business_id"]))
elif action == "get_service":
return run_async(self.get_service(params["business_id"], params["item_id"]))
elif action == "list_staff":
return run_async(self.list_staff(params["business_id"]))
elif action == "get_staff":
return run_async(self.get_staff(params["business_id"], params["item_id"]))
elif action == "list_appointments":
return run_async(self.list_appointments(
params["business_id"],
params.get("start_date"),
params.get("end_date")
))
elif action == "get_appointment":
return run_async(self.get_appointment(params["business_id"], params["item_id"]))
elif action == "create_appointment":
return run_async(self.create_appointment(params["business_id"], params))
elif action == "update_appointment":
return run_async(self.update_appointment(
params["business_id"],
params["appointment_id"],
params
))
elif action == "cancel_appointment":
return run_async(self.cancel_appointment(
params["business_id"],
params["appointment_id"]
))
elif action == "list_customers":
return run_async(self.list_customers(params["business_id"]))
elif action == "get_customer":
return run_async(self.get_customer(params["business_id"], params["customer_id"]))
# Hybrid Booking Workflow actions (delegated to hybrid_ops)
elif action == "create_hybrid_session":
return run_async(self.hybrid_ops.create_hybrid_session(
organizer_email=params["organizer_email"],
organizer_name=params["organizer_name"],
internal_attendees=params.get("internal_attendees", []),
external_email=params["external_email"],
external_name=params.get("external_name"),
proposed_slots=params["proposed_slots"],
meeting_subject=params.get("meeting_subject"),
meeting_duration=params.get("meeting_duration", 30)
))
elif action == "send_hybrid_invitation":
return run_async(self.hybrid_ops.send_hybrid_invitation(params["session_id"]))
elif action == "get_hybrid_session":
return run_async(self.hybrid_ops.get_hybrid_session(params["session_id"]))
elif action == "check_hybrid_slot":
return run_async(self.hybrid_ops.check_hybrid_slot(
params["session_id"],
params["slot_index"]
))
elif action == "confirm_hybrid_booking":
return run_async(self.hybrid_ops.confirm_hybrid_booking(
params["session_id"],
params["selected_slot_index"]
))
elif action == "find_new_hybrid_slots":
return run_async(self.hybrid_ops.find_new_hybrid_slots(params["session_id"]))
# ============================================================================
# Microsoft Bookings API Methods
# ============================================================================
async def list_businesses(self) -> Dict[str, Any]:
"""List all Microsoft Bookings businesses accessible to the user"""
try:
response = await self.client.get("/solutions/bookingBusinesses")
businesses = response.get("value", [])
return {
"businesses": businesses,
"count": len(businesses)
}
except Exception as e:
logger.error(f"Failed to list businesses: {e}")
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list businesses: {str(e)}"
)
async def get_business(self, business_id: str) -> Dict[str, Any]:
"""Get details of a specific Bookings business"""
try:
business = await self.client.get(f"/solutions/bookingBusinesses/{business_id}")
return {"business": business}
except Exception as e:
logger.error(f"Failed to get business: {e}")
raise OperationError(
code="GET_FAILED",
message=f"Failed to get business: {str(e)}"
)
async def list_services(self, business_id: str) -> Dict[str, Any]:
"""List all services in a Bookings business"""
try:
response = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/services"
)
services = response.get("value", [])
return {
"services": services,
"count": len(services)
}
except Exception as e:
logger.error(f"Failed to list services: {e}")
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list services: {str(e)}"
)
async def get_service(self, business_id: str, service_id: str) -> Dict[str, Any]:
"""Get service details"""
try:
service = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/services/{service_id}"
)
return {"service": service}
except Exception as e:
logger.error(f"Failed to get service: {e}")
raise OperationError(
code="GET_FAILED",
message=f"Failed to get service: {str(e)}"
)
async def list_staff(self, business_id: str) -> Dict[str, Any]:
"""List all staff members in a Bookings business"""
try:
response = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/staffMembers"
)
staff = response.get("value", [])
return {
"staff": staff,
"count": len(staff)
}
except Exception as e:
logger.error(f"Failed to list staff: {e}")
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list staff: {str(e)}"
)
async def get_staff(self, business_id: str, staff_id: str) -> Dict[str, Any]:
"""Get staff member details"""
try:
staff = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/staffMembers/{staff_id}"
)
return {"staff": staff}
except Exception as e:
logger.error(f"Failed to get staff: {e}")
raise OperationError(
code="GET_FAILED",
message=f"Failed to get staff: {str(e)}"
)
async def list_appointments(
self,
business_id: str,
start_date: str = None,
end_date: str = None
) -> Dict[str, Any]:
"""List appointments with optional date filters"""
try:
params = {}
if start_date and end_date:
params["$filter"] = f"start/dateTime ge '{start_date}' and end/dateTime le '{end_date}'"
response = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/appointments",
params=params
)
appointments = response.get("value", [])
return {
"appointments": appointments,
"count": len(appointments)
}
except Exception as e:
logger.error(f"Failed to list appointments: {e}")
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list appointments: {str(e)}"
)
async def get_appointment(self, business_id: str, appointment_id: str) -> Dict[str, Any]:
"""Get appointment details"""
try:
appointment = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/appointments/{appointment_id}"
)
return {"appointment": appointment}
except Exception as e:
logger.error(f"Failed to get appointment: {e}")
raise OperationError(
code="GET_FAILED",
message=f"Failed to get appointment: {str(e)}"
)
async def create_appointment(
self,
business_id: str,
appointment_data: Dict[str, Any]
) -> Dict[str, Any]:
"""Create a new appointment"""
try:
created = await self.client.post(
f"/solutions/bookingBusinesses/{business_id}/appointments",
json=appointment_data
)
return {"appointment": created}
except Exception as e:
logger.error(f"Failed to create appointment: {e}")
raise OperationError(
code="CREATE_FAILED",
message=f"Failed to create appointment: {str(e)}"
)
async def update_appointment(
self,
business_id: str,
appointment_id: str,
update_data: Dict[str, Any]
) -> Dict[str, Any]:
"""Update an existing appointment"""
try:
updated = await self.client.patch(
f"/solutions/bookingBusinesses/{business_id}/appointments/{appointment_id}",
json=update_data
)
return {"appointment": updated}
except Exception as e:
logger.error(f"Failed to update appointment: {e}")
raise OperationError(
code="UPDATE_FAILED",
message=f"Failed to update appointment: {str(e)}"
)
async def cancel_appointment(self, business_id: str, appointment_id: str) -> Dict[str, Any]:
"""Cancel an appointment"""
try:
await self.client.delete(
f"/solutions/bookingBusinesses/{business_id}/appointments/{appointment_id}"
)
return {"success": True, "appointment_id": appointment_id}
except Exception as e:
logger.error(f"Failed to cancel appointment: {e}")
raise OperationError(
code="DELETE_FAILED",
message=f"Failed to cancel appointment: {str(e)}"
)
async def list_customers(self, business_id: str) -> Dict[str, Any]:
"""List all customers in a Bookings business"""
try:
response = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/customers"
)
customers = response.get("value", [])
return {
"customers": customers,
"count": len(customers)
}
except Exception as e:
logger.error(f"Failed to list customers: {e}")
raise OperationError(
code="LIST_FAILED",
message=f"Failed to list customers: {str(e)}"
)
async def get_customer(self, business_id: str, customer_id: str) -> Dict[str, Any]:
"""Get customer details"""
try:
customer = await self.client.get(
f"/solutions/bookingBusinesses/{business_id}/customers/{customer_id}"
)
return {"customer": customer}
except Exception as e:
logger.error(f"Failed to get customer: {e}")
raise OperationError(
code="GET_FAILED",
message=f"Failed to get customer: {str(e)}"
)
