async def close(self):
"""Close the HTTP client."""
await self.client.aclose()
"""Canvas API client for interacting with Canvas LMS."""
import os
from typing import Dict, List, Any, Optional
import httpx
from datetime import datetime, timedelta
class CanvasAPI:
"""Canvas API client for CMU Canvas instance."""
def __init__(self, base_url: str = "https://canvas.cmu.edu", api_token: Optional[str] = None):
self.base_url = base_url
self.api_token = api_token or os.getenv("CANVAS_API_TOKEN")
if not self.api_token:
raise ValueError("Canvas API token is required. Set CANVAS_API_TOKEN environment variable.")
self.client = httpx.AsyncClient(timeout=30.0)
async def _make_request(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Any:
"""Make an authenticated request to the Canvas API."""
url = f"{self.base_url}/api/v1{endpoint}"
# Add access token to params
if params is None:
params = {}
params["access_token"] = self.api_token
try:
response = await self.client.get(url, params=params)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
raise Exception(f"Canvas API request failed: {e}")
async def get_user_profile(self) -> Dict[str, Any]:
"""Get the current user's profile information."""
return await self._make_request("/users/self")
async def get_courses(self, enrollment_state: str = "active") -> List[Dict[str, Any]]:
"""Get user's enrolled courses."""
params = {"enrollment_state": enrollment_state}
courses = await self._make_request("/courses", params)
# Filter to only courses with enrollment_state=active
active_courses = []
for course in courses:
enrollments = course.get("enrollments", [])
if enrollments and enrollments[0].get("enrollment_state") == "active":
active_courses.append(course)
return active_courses
async def get_upcoming_events(self, days_ahead: int = 30) -> List[Dict[str, Any]]:
"""Get upcoming assignments and events."""
events = await self._make_request("/users/self/upcoming_events")
# Filter to events within the specified timeframe
cutoff_date = datetime.now() + timedelta(days=days_ahead)
upcoming_assignments = []
for event in events:
# Get event date from assignment due date or event start date
event_date_str = None
if event.get("assignment", {}).get("due_at"):
event_date_str = event["assignment"]["due_at"]
elif event.get("start_at"):
event_date_str = event["start_at"]
if event_date_str:
try:
# Parse ISO format date
event_date = datetime.fromisoformat(event_date_str.replace("Z", "+00:00"))
if event_date <= cutoff_date:
# Only include assignments and quizzes
if event.get("assignment") or event.get("type") == "assignment":
upcoming_assignments.append(event)
except ValueError:
continue
return upcoming_assignments
async def get_course_assignments(self, course_id: str) -> List[Dict[str, Any]]:
"""Get all assignments for a specific course."""
return await self._make_request(f"/courses/{course_id}/assignments")
async def get_todo_items(self) -> List[Dict[str, Any]]:
"""Get user's To-Do list items."""
return await self._make_request("/users/self/todo")
async def get_assignment_details(self, course_id: str, assignment_id: str) -> Dict[str, Any]:
"""Get detailed information about a specific assignment."""
return await self._make_request(f"/courses/{course_id}/assignments/{assignment_id}")
async def get_grades(self, course_id: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get grades for all courses or a specific course."""
if course_id:
# Get assignments with grades for specific course
assignments = await self._make_request(f"/courses/{course_id}/assignments",
{"include": ["submission"]})
return assignments
else:
# Get grades across all courses
enrollments = await self._make_request("/users/self/enrollments",
{"include": ["current_grading_period_scores", "total_scores"]})
return enrollments
async def get_calendar_events(self, start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get calendar events within a date range."""
params = {"type": "event"}
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
return await self._make_request("/calendar_events", params)
async def get_discussion_topics(self, course_id: str) -> List[Dict[str, Any]]:
"""Get discussion topics for a course."""
return await self._make_request(f"/courses/{course_id}/discussion_topics")
async def get_announcements(self, course_id: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get announcements for a course or all courses."""
if course_id:
return await self._make_request(f"/courses/{course_id}/discussion_topics",
{"only_announcements": "true"})
else:
# Get announcements across all courses
return await self._make_request("/announcements")
async def get_assignment_submissions(self, course_id: str, assignment_id: str) -> List[Dict[str, Any]]:
"""Get submissions for a specific assignment."""
return await self._make_request(f"/courses/{course_id}/assignments/{assignment_id}/submissions/self")
async def get_course_modules(self, course_id: str) -> List[Dict[str, Any]]:
"""Get course modules and their items."""
return await self._make_request(f"/courses/{course_id}/modules",
{"include": ["items", "content_details"]})
async def get_course_activity_stream(self) -> List[Dict[str, Any]]:
"""Get recent activity across all courses."""
return await self._make_request("/users/self/activity_stream")
