How to Integrate Replit with Canvas LMS

1// Verify Canvas secrets are set before starting server
2const required = ['CANVAS_API_TOKEN', 'CANVAS_BASE_URL'];
3for (const key of required) {
4  if (!process.env[key]) {
5    throw new Error(`Missing secret: ${key}. Set it in Replit Secrets (lock icon 🔒).`);
6  }
7}
8console.log('Canvas config OK. Base URL:', process.env.CANVAS_BASE_URL);

1// canvas.js — Canvas LMS REST API integration for Node.js on Replit
2const axios = require('axios');
3const express = require('express');
4
5const BASE_URL = process.env.CANVAS_BASE_URL;
6const TOKEN = process.env.CANVAS_API_TOKEN;
7
8const canvasApi = axios.create({
9  baseURL: `${BASE_URL}/api/v1`,
10  headers: { Authorization: `Bearer ${TOKEN}` }
11});
12
13// Follow Canvas pagination to get all results
14async function getAllPages(url, params = {}) {
15  const results = [];
16  let nextUrl = url;
17  let nextParams = { ...params, per_page: 100 };
18  
19  while (nextUrl) {
20    const response = await canvasApi.get(nextUrl, { params: nextParams });
21    results.push(...response.data);
22    
23    // Parse Link header for next page URL
24    const linkHeader = response.headers['link'] || '';
25    const nextMatch = linkHeader.match(/<([^>]+)>;\s*rel="next"/);
26    nextUrl = nextMatch ? nextMatch[1] : null;
27    nextParams = {};  // URL already includes params after first page
28  }
29  
30  return results;
31}
32
33const app = express();
34app.use(express.json());
35
36// List all courses for the authenticated user
37app.get('/api/courses', async (req, res) => {
38  try {
39    const courses = await getAllPages('/courses', { enrollment_type: 'teacher' });
40    res.json({ courses: courses.map(c => ({ id: c.id, name: c.name, code: c.course_code })) });
41  } catch (err) {
42    res.status(err.response?.status || 500).json({ error: err.message });
43  }
44});
45
46// Get all enrollments for a course
47app.get('/api/courses/:courseId/students', async (req, res) => {
48  try {
49    const enrollments = await getAllPages(`/courses/${req.params.courseId}/enrollments`, {
50      type: ['StudentEnrollment'],
51      state: ['active']
52    });
53    const students = enrollments.map(e => ({
54      id: e.user_id,
55      name: e.user?.name,
56      email: e.user?.login_id
57    }));
58    res.json({ students, total: students.length });
59  } catch (err) {
60    res.status(err.response?.status || 500).json({ error: err.message });
61  }
62});
63
64// Get assignments for a course
65app.get('/api/courses/:courseId/assignments', async (req, res) => {
66  try {
67    const assignments = await getAllPages(`/courses/${req.params.courseId}/assignments`);
68    res.json({
69      assignments: assignments.map(a => ({
70        id: a.id,
71        name: a.name,
72        due_at: a.due_at,
73        points_possible: a.points_possible
74      }))
75    });
76  } catch (err) {
77    res.status(err.response?.status || 500).json({ error: err.message });
78  }
79});
80
81// Post a grade for a student submission
82app.put('/api/courses/:courseId/assignments/:assignmentId/grade', async (req, res) => {
83  const { courseId, assignmentId } = req.params;
84  const { studentId, score, comment } = req.body;
85  try {
86    const response = await canvasApi.put(
87      `/courses/${courseId}/assignments/${assignmentId}/submissions/${studentId}`,
88      {
89        submission: { posted_grade: score },
90        comment: comment ? { text_comment: comment } : undefined
91      }
92    );
93    res.json({ success: true, submission: response.data });
94  } catch (err) {
95    res.status(err.response?.status || 500).json({ error: err.message });
96  }
97});
98
99app.listen(3000, '0.0.0.0', () => console.log('Canvas server running on port 3000'));

1# canvas_api.py — Canvas LMS REST API integration for Python on Replit
2import os
3import requests
4from flask import Flask, request, jsonify
5
6BASE_URL = os.environ['CANVAS_BASE_URL']
7TOKEN = os.environ['CANVAS_API_TOKEN']
8
9# Create a session with Canvas auth header
10session = requests.Session()
11session.headers.update({'Authorization': f'Bearer {TOKEN}'})
12
13def get_all_pages(path: str, params: dict = None) -> list:
14    """Fetch all pages from a paginated Canvas API endpoint."""
15    results = []
16    url = f'{BASE_URL}/api/v1{path}'
17    req_params = {'per_page': 100, **(params or {})}
18    
19    while url:
20        response = session.get(url, params=req_params)
21        response.raise_for_status()
22        results.extend(response.json())
23        
24        # Follow Link header for next page
25        link_header = response.headers.get('Link', '')
26        next_url = None
27        for part in link_header.split(','):
28            if 'rel="next"' in part:
29                next_url = part.split(';')[0].strip().strip('<>')
30                break
31        url = next_url
32        req_params = {}  # Params embedded in next URL
33    
34    return results
35
36def post_grade(course_id: int, assignment_id: int, student_id: int, score, comment: str = None):
37    """Post a grade for a student submission."""
38    url = f'{BASE_URL}/api/v1/courses/{course_id}/assignments/{assignment_id}/submissions/{student_id}'
39    payload = {'submission': {'posted_grade': score}}
40    if comment:
41        payload['comment'] = {'text_comment': comment}
42    response = session.put(url, json=payload)
43    response.raise_for_status()
44    return response.json()
45
46app = Flask(__name__)
47
48@app.route('/api/courses')
49def list_courses():
50    try:
51        courses = get_all_pages('/courses', {'enrollment_type': 'teacher'})
52        return jsonify({'courses': [{'id': c['id'], 'name': c['name']} for c in courses]})
53    except requests.HTTPError as e:
54        return jsonify({'error': str(e)}), e.response.status_code
55
56@app.route('/api/courses/<int:course_id>/students')
57def list_students(course_id):
58    try:
59        enrollments = get_all_pages(f'/courses/{course_id}/enrollments',
60                                    {'type[]': 'StudentEnrollment', 'state[]': 'active'})
61        students = [{'id': e['user_id'], 'name': e.get('user', {}).get('name')} for e in enrollments]
62        return jsonify({'students': students, 'total': len(students)})
63    except requests.HTTPError as e:
64        return jsonify({'error': str(e)}), e.response.status_code
65
66@app.route('/api/grade', methods=['POST'])
67def grade_submission():
68    data = request.get_json()
69    try:
70        result = post_grade(
71            data['course_id'], data['assignment_id'],
72            data['student_id'], data['score'],
73            data.get('comment')
74        )
75        return jsonify({'success': True, 'submission': result})
76    except requests.HTTPError as e:
77        return jsonify({'error': str(e)}), e.response.status_code
78
79if __name__ == '__main__':
80    app.run(host='0.0.0.0', port=3000)