nav[aria-label="Primary Navigation"] { padding: 0; & ul { list-style: none; width: 100%; display: flex; flex-direction: row; justify-content: start; align-items: start; gap: 30px; padding: 0; & li { margin: 0; } & ul li { list-style: none; } } }

Bulk Upload Student Photos to Canvas LMS

GloriaRamos58664818

I have never used Powershell. Can someone provide me with step by step process of implementing this to bulk upload photos??

Find more posts tagged with

Canvas LMS

SIS

Accepted answers

GloriaRamos58664818

Chris,

Thank you so much for reaching out.

1- yes

2- in my google drive; named with their school id

3- no pictures exist in Canvas

4- I am open to any solution other than uploading one at a time.

All comments

Chris

Hi @GloriaRamos58664818,

I feel like this may end up being a more complicated question than is seems. Could you answer a few questions that may help to get you the best answer possible?

1 - Are you a Canvas administrator?

2 - Where are the student photos currently located? If they are on a local computer, are they named in an identifiable way (like the studnt's sis_id or login_id)?

3 - Are you looking to replace any profile pictures that already exist in Canvas?

4 - Is Powershell the only option you can use, or would you be open to alternatives like Python?

-Chris

GloriaRamos58664818

Chris,

Thank you so much for reaching out.

1- yes

2- in my google drive; named with their school id

3- no pictures exist in Canvas

4- I am open to any solution other than uploading one at a time.

Chris

Hi Gloria,

You could try this python script (use your beta or test Canvas instance first to ensure it works before trying in production). It's generated largely form claude ai, and assumes the names of the files match a user's sis+id in Canvas.

#!/usr/bin/env python3
"""
Bulk-upload student profile pictures to Instructure Canvas LMS.

Files in the source folder must be named after the student's SIS user ID,
e.g. 900123456.jpg -> Canvas user sis_user_id:900123456

Canvas has no single "set avatar" endpoint. For each student this script:
  1. POST /api/v1/users/:id/files            -> get a pre-signed upload target
  2. POST <upload_url>                        -> send the actual bytes
  3. GET  /api/v1/users/:id/avatars           -> find the token for that upload
  4. PUT  /api/v1/users/:id                   -> set user[avatar][token]

Usage:
    export CANVAS_TOKEN="your_admin_token"
    python canvas_bulk_avatars.py \
        --base-url https://yourschool.instructure.com \
        --folder ./student_photos \
        --dry-run

    # then run for real
    python canvas_bulk_avatars.py \
        --base-url https://yourschool.instructure.com \
        --folder ./student_photos \
        --workers 4 \
        --log results.csv
"""

import argparse
import csv
import mimetypes
import os
import random
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import requests

IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
AVATAR_FOLDER = "profile pictures"  # Canvas' magic folder name; don't rename it
MAX_RETRIES = 5
_print_lock = threading.Lock()


def log(msg):
    with _print_lock:
        print(msg, flush=True)


class CanvasClient:
    def __init__(self, base_url, token, masquerade=True, timeout=60):
        self.base = base_url.rstrip("/")
        self.timeout = timeout
        self.masquerade = masquerade
        self.session = requests.Session()
        self.session.headers["Authorization"] = f"Bearer {token}"

    def request(self, method, path, as_user=None, **kwargs):
        """Call the Canvas API with retry on rate limiting / transient errors."""
        url = path if path.startswith("http") else f"{self.base}{path}"

        if as_user and self.masquerade:
            params = dict(kwargs.pop("params", None) or {})
            params["as_user_id"] = as_user
            kwargs["params"] = params

        for attempt in range(MAX_RETRIES):
            resp = self.session.request(method, url, timeout=self.timeout, **kwargs)

            # Canvas signals throttling with 403 + a plain-text body
            throttled = resp.status_code == 429 or (
                resp.status_code == 403 and "Rate Limit Exceeded" in resp.text
            )
            if throttled or resp.status_code in (500, 502, 503, 504):
                if attempt == MAX_RETRIES - 1:
                    return resp
                wait = (2 ** attempt) + random.uniform(0, 1)
                log(f"    throttled/{resp.status_code}, retrying in {wait:.1f}s")
                time.sleep(wait)
                continue
            return resp
        return resp

    def get_user(self, sis_id):
        """Resolve a SIS ID to the Canvas user record."""
        resp = self.request("GET", f"/api/v1/users/sis_user_id:{sis_id}")
        if resp.status_code == 404:
            raise LookupError(f"no Canvas user with sis_user_id {sis_id}")
        resp.raise_for_status()
        return resp.json()

    def upload_avatar_file(self, user_id, filepath, as_user):
        """Steps 1 and 2: push the image into the user's profile pictures folder."""
        filepath = Path(filepath)
        size = filepath.stat().st_size
        content_type = mimetypes.guess_type(filepath.name)[0] or "image/jpeg"

        # Step 1 -- tell Canvas what's coming and get an upload target back.
        init = self.request(
            "POST",
            f"/api/v1/users/{user_id}/files",
            as_user=as_user,
            data={
                "name": filepath.name,
                "size": size,
                "content_type": content_type,
                "parent_folder_path": AVATAR_FOLDER,
                "on_duplicate": "overwrite",
                "no_redirect": True,
            },
        )
        if not init.ok:
            raise RuntimeError(f"file init failed ({init.status_code}): {init.text[:300]}")
        init_data = init.json()

        upload_url = init_data["upload_url"]
        upload_params = init_data.get("upload_params", {})

        # Step 2 -- POST to the storage backend. No auth header here: the URL is
        # pre-signed, and sending credentials to S3 will make it reject the request.
        with filepath.open("rb") as fh:
            files = {"file": (filepath.name, fh, content_type)}
            upload = requests.post(
                upload_url,
                data=upload_params,
                files=files,
                allow_redirects=False,
                timeout=self.timeout,
            )

        if upload.status_code in (301, 302, 303):
            # Older/S3-backed instances hand back a redirect that must be
            # confirmed with our Canvas credentials.
            confirm = self.request("GET", upload.headers["Location"])
            confirm.raise_for_status()
            return confirm.json()

        if not upload.ok:
            raise RuntimeError(f"upload failed ({upload.status_code}): {upload.text[:300]}")
        return upload.json()

    def set_avatar(self, user_id, attachment, as_user):
        """Steps 3 and 4: find the token for the new image and apply it."""
        resp = self.request("GET", f"/api/v1/users/{user_id}/avatars", as_user=as_user)
        resp.raise_for_status()
        options = resp.json()

        token = None
        for opt in options:
            if opt.get("type") != "attachment":
                continue
            # Match on the attachment we just created; fall back to display name.
            if str(opt.get("id", "")).endswith(str(attachment.get("id"))) or opt.get(
                "display_name"
            ) == attachment.get("display_name"):
                token = opt.get("token")
                break
        if token is None:
            for opt in options:
                if opt.get("type") == "attachment" and opt.get("token"):
                    token = opt["token"]
                    break
        if token is None:
            raise RuntimeError("uploaded image did not appear in the user's avatar options")

        update = self.request(
            "PUT",
            f"/api/v1/users/{user_id}",
            as_user=as_user,
            data={"user[avatar][token]": token},
        )
        if not update.ok:
            raise RuntimeError(f"avatar update failed ({update.status_code}): {update.text[:300]}")
        return update.json()


def process(client, filepath, dry_run=False):
    sis_id = Path(filepath).stem.strip()
    as_user = f"sis_user_id:{sis_id}"

    user = client.get_user(sis_id)
    user_id = user["id"]
    name = user.get("sortable_name") or user.get("name", "")

    if dry_run:
        return f"would set avatar for {sis_id} ({name}, canvas id {user_id})"

    attachment = client.upload_avatar_file(user_id, filepath, as_user)
    client.set_avatar(user_id, attachment, as_user)
    return f"set avatar for {sis_id} ({name})"


def collect_images(folder):
    files = [
        p for p in sorted(Path(folder).iterdir())
        if p.is_file() and p.suffix.lower() in IMAGE_EXTS
    ]
    if not files:
        sys.exit(f"No image files found in {folder}")
    return files


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--base-url", required=True, help="e.g. https://yourschool.instructure.com")
    ap.add_argument("--folder", required=True, help="folder of images named <sis_user_id>.jpg")
    ap.add_argument("--token", default=os.environ.get("CANVAS_TOKEN"),
                    help="Canvas API token (defaults to $CANVAS_TOKEN)")
    ap.add_argument("--workers", type=int, default=1,
                    help="parallel uploads; keep low to avoid rate limits (default 1)")
    ap.add_argument("--limit", type=int, help="only process the first N files (for testing)")
    ap.add_argument("--dry-run", action="store_true",
                    help="resolve every SIS ID but upload nothing")
    ap.add_argument("--no-masquerade", action="store_true",
                    help="don't send as_user_id; only works if the token already owns the files")
    ap.add_argument("--log", help="write a CSV report of results here")
    args = ap.parse_args()

    if not args.token:
        sys.exit("No API token. Pass --token or set CANVAS_TOKEN.")

    files = collect_images(args.folder)
    if args.limit:
        files = files[: args.limit]

    client = CanvasClient(args.base_url, args.token, masquerade=not args.no_masquerade)
    log(f"{len(files)} image(s) to process{' (dry run)' if args.dry_run else ''}\n")

    results = []

    def run_one(path):
        try:
            return path, "ok", process(client, path, args.dry_run)
        except Exception as exc:
            return path, "error", f"{type(exc).__name__}: {exc}"

    if args.workers > 1:
        with ThreadPoolExecutor(max_workers=args.workers) as pool:
            futures = [pool.submit(run_one, f) for f in files]
            for fut in as_completed(futures):
                results.append(fut.result())
                path, status, msg = results[-1]
                log(f"[{status:5}] {path.name}: {msg}")
    else:
        for i, path in enumerate(files, 1):
            result = run_one(path)
            results.append(result)
            _, status, msg = result
            log(f"[{i}/{len(files)}] [{status:5}] {path.name}: {msg}")

    ok = sum(1 for _, s, _ in results if s == "ok")
    log(f"\nDone. {ok} succeeded, {len(results) - ok} failed.")

    if args.log:
        with open(args.log, "w", newline="", encoding="utf-8") as fh:
            writer = csv.writer(fh)
            writer.writerow(["file", "sis_user_id", "status", "detail"])
            for path, status, msg in sorted(results, key=lambda r: r[0].name):
                writer.writerow([path.name, path.stem, status, msg])
        log(f"Report written to {args.log}")

    sys.exit(1 if ok < len(results) else 0)


if __name__ == "__main__":
    main()

-Chris

GloriaRamos58664818

Thank You!

Unanswered Questions

Solicitud de apoyo – Error al modificar opciones de respuesta en New Quizzes
Cordial saludo Actualmente estamos presentando una inconsistencia en New Quizzes al momento de realizar la migración y posterior edición de las evaluaciones. Específicamente, al intentar modificar una pregunta y eliminar una de sus opciones de respuesta, el sistema no permite ejecutar dicha acción. Con el fin de descartar…
Rubric with letter grade while Gradebook shows points
Is there a way for me to create a rubric that shows BOTH points and a letter grade, while the Gradebook in both my view and student view shows JUST points? I tried setting up the rubric showing a letter grade in the grading scheme, and a letter shows in the Assessment box…but then my gradebook also shows the letter grade,…
Canvas Outcomes – Best Way to Structure Multiple Skill Sets for Reporting?
Our institution is setting up 9 institutional outcomes in Canvas. Several of our outcomes have multiple skill sets within the rubric. For example: Critical Reading – Skill Set A Critical Reading – Skill Set B Critical Reading – Skill Set C Our understanding is that Canvas Outcomes data reports can provide the overall/total…
Outcome rating descriptions
I am just curiousas to why there is not a field to enter a description for an outcome at the account level? Or at least edit the rating when you attach/link an account level outcome to a rubric. When you create a new criterion for a rubric, you enter a decription for a rating, but for an account level outcome.
Guaranteed Order of Analysis CSV Export File in New Quizzes
We use a pre- post- model to measure learning which requires student-question matching from two time periods. This requires exporting quiz data to CSV to do the analysis properly. (I can get into more detail if needed.) The New Quizzes Student Analysis CSV export has repeated blocks of ItemID then columns related to that…