Function check_course_access

apps/api/src/services/admin/admin.py:195–228

Check if a user can access a specific course within the token's org.

(
    token_user: APITokenUser,
    course_uuid: str,
    user_id: int,
    db_session: AsyncSession,
)

async def check_course_access(
    token_user: APITokenUser,
    course_uuid: str,
    user_id: int,
    db_session: AsyncSession,
) -> dict:
    """Check if a user can access a specific course within the token's org."""

    await _get_user_in_org(user_id, token_user.org_id, db_session)

    course = (await db_session.execute(
        select(Course).where(
            Course.course_uuid == course_uuid,
            Course.org_id == token_user.org_id,
        )
    )).scalars().first()
    if not course:
        raise HTTPException(status_code=404, detail="Course not found")

    # Check if user has a TrailRun (enrollment) for this course
    enrollment = (await db_session.execute(
        select(TrailRun).where(
            TrailRun.course_id == course.id,
            TrailRun.user_id == user_id,
            TrailRun.org_id == token_user.org_id,
        )
    )).scalars().first()

    return {
        "has_access": course.public or enrollment is not None,
        "is_enrolled": enrollment is not None,
        "is_public": course.public,
        "is_published": course.published,
    }


# -- Enrollment endpoints -----------------------------------------------------