coldcall_lti/coldcall_lti/launch.py

"""Turn validated LTI launch claims into database rows.

Kept separate from the Flask/pylti1p3 plumbing so it can be tested with
plain dicts of claims and reused by the dev-mode fake launcher.
"""

import datetime
from dataclasses import dataclass

from sqlalchemy import select

from .models import Course, Enrollment, Student, ensure_default_levels

CLAIM_CONTEXT = "https://purl.imsglobal.org/spec/lti/claim/context"
CLAIM_ROLES = "https://purl.imsglobal.org/spec/lti/claim/roles"
CLAIM_DEPLOYMENT = "https://purl.imsglobal.org/spec/lti/claim/deployment_id"
CLAIM_CUSTOM = "https://purl.imsglobal.org/spec/lti/claim/custom"
CLAIM_RESOURCE_LINK = "https://purl.imsglobal.org/spec/lti/claim/resource_link"


def clean_custom_value(value):
    """Custom parameters echo back their '$...' variable name when the
    platform doesn't support the expansion; treat that as absent."""
    if not value or str(value).startswith("$"):
        return None
    return value


def parse_custom_date(value):
    """Canvas date expansions arrive as ISO-ish timestamps; keep the
    date part and tolerate absence or junk."""
    value = clean_custom_value(value)
    if not value:
        return None
    try:
        return datetime.date.fromisoformat(str(value)[:10])
    except ValueError:
        return None

ROLE_INSTRUCTOR = "http://purl.imsglobal.org/vocab/lis/v2/membership#Instructor"
ROLE_LEARNER = "http://purl.imsglobal.org/vocab/lis/v2/membership#Learner"


@dataclass
class LaunchInfo:
    context_id: str
    deployment_id: str | None
    course_title: str | None
    user_id: str
    name: str | None
    email: str | None
    picture: str | None
    pronouns: str | None
    resource_link_id: str | None
    course_start: datetime.date | None
    course_end: datetime.date | None
    is_instructor: bool


def roles_include_instructor(roles):
    return any("membership#Instructor" in r for r in roles)


def extract_launch_info(launch_data):
    context = launch_data.get(CLAIM_CONTEXT, {})
    custom = launch_data.get(CLAIM_CUSTOM, {})
    return LaunchInfo(
        context_id=context["id"],
        deployment_id=launch_data.get(CLAIM_DEPLOYMENT),
        course_title=context.get("title"),
        user_id=launch_data["sub"],
        name=launch_data.get("name"),
        email=launch_data.get("email"),
        picture=launch_data.get("picture"),
        pronouns=clean_custom_value(custom.get("pronouns")),
        resource_link_id=launch_data.get(CLAIM_RESOURCE_LINK, {}).get("id"),
        course_start=parse_custom_date(custom.get("course_start")),
        course_end=parse_custom_date(custom.get("course_end")),
        is_instructor=roles_include_instructor(launch_data.get(CLAIM_ROLES, [])),
    )


def upsert_launch(db, info):
    """Create or refresh the course, user, and enrollment for a launch."""
    course = db.execute(
        select(Course).where(Course.lti_context_id == info.context_id)
    ).scalar_one_or_none()
    if course is None:
        course = Course(lti_context_id=info.context_id)
        db.add(course)
        db.flush()
        ensure_default_levels(db, course)
    if info.course_title:
        course.title = info.course_title
    if info.deployment_id:
        course.lti_deployment_id = info.deployment_id
    if info.course_start:
        course.start_date = info.course_start
    if info.course_end:
        course.end_date = info.course_end

    student = db.execute(
        select(Student).where(Student.canvas_user_id == info.user_id)
    ).scalar_one_or_none()
    if student is None:
        student = Student(canvas_user_id=info.user_id)
        db.add(student)
    if info.name:
        student.name = info.name
    if info.email:
        student.email = info.email
    if info.picture:
        student.avatar_url = info.picture
    if info.pronouns:
        student.pronouns = info.pronouns
    db.flush()

    enrollment = db.execute(
        select(Enrollment).where(
            Enrollment.course_id == course.id,
            Enrollment.student_id == student.id,
        )
    ).scalar_one_or_none()
    role = "instructor" if info.is_instructor else "student"
    if enrollment is None:
        enrollment = Enrollment(
            course_id=course.id, student_id=student.id, role=role
        )
        db.add(enrollment)
    else:
        enrollment.role = role
        enrollment.active = True
    db.flush()

    return course, student