bc30addfb9
fixing unit tests fixing merge error fixing xqueue submission issue with unicode url (trial 0.1) fixing fotmats as commented upon removing yaml file language selection Unicode changes to support QRF removed unnecessary pass in modulestore/init.py fixing merge error fixing fotmats as commented upon removing yaml file language selection fixing pep8 violations - fixing pylint violations pylint violation fixing line spaces and formats ignore pylint E1101 remove empty line fixing pylint violations pep8 violations bulk mail unicode/decode fix migration error fix pep8 just to push again more unicode/decode Final changes to comments and error messages.
277 lines
8.1 KiB
Python
277 lines
8.1 KiB
Python
"""
|
|
This file contains the logic for cohort groups, as exposed internally to the
|
|
forums, and to the cohort admin views.
|
|
"""
|
|
|
|
from django.http import Http404
|
|
import logging
|
|
import random
|
|
|
|
from courseware import courses
|
|
from student.models import get_user_by_username_or_email
|
|
from .models import CourseUserGroup
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# tl;dr: global state is bad. capa reseeds random every time a problem is loaded. Even
|
|
# if and when that's fixed, it's a good idea to have a local generator to avoid any other
|
|
# code that messes with the global random module.
|
|
_local_random = None
|
|
|
|
def local_random():
|
|
"""
|
|
Get the local random number generator. In a function so that we don't run
|
|
random.Random() at import time.
|
|
"""
|
|
# ironic, isn't it?
|
|
global _local_random
|
|
|
|
if _local_random is None:
|
|
_local_random = random.Random()
|
|
|
|
return _local_random
|
|
|
|
def is_course_cohorted(course_id):
|
|
"""
|
|
Given a course id, return a boolean for whether or not the course is
|
|
cohorted.
|
|
|
|
Raises:
|
|
Http404 if the course doesn't exist.
|
|
"""
|
|
return courses.get_course_by_id(course_id).is_cohorted
|
|
|
|
|
|
def get_cohort_id(user, course_id):
|
|
"""
|
|
Given a course id and a user, return the id of the cohort that user is
|
|
assigned to in that course. If they don't have a cohort, return None.
|
|
"""
|
|
cohort = get_cohort(user, course_id)
|
|
return None if cohort is None else cohort.id
|
|
|
|
|
|
def is_commentable_cohorted(course_id, commentable_id):
|
|
"""
|
|
Args:
|
|
course_id: string
|
|
commentable_id: string
|
|
|
|
Returns:
|
|
Bool: is this commentable cohorted?
|
|
|
|
Raises:
|
|
Http404 if the course doesn't exist.
|
|
"""
|
|
course = courses.get_course_by_id(course_id)
|
|
|
|
if not course.is_cohorted:
|
|
# this is the easy case :)
|
|
ans = False
|
|
elif commentable_id in course.top_level_discussion_topic_ids:
|
|
# top level discussions have to be manually configured as cohorted
|
|
# (default is not)
|
|
ans = commentable_id in course.cohorted_discussions
|
|
else:
|
|
# inline discussions are cohorted by default
|
|
ans = True
|
|
|
|
log.debug(u"is_commentable_cohorted({0}, {1}) = {2}".format(course_id,
|
|
commentable_id,
|
|
ans))
|
|
return ans
|
|
|
|
|
|
def get_cohorted_commentables(course_id):
|
|
"""
|
|
Given a course_id return a list of strings representing cohorted commentables
|
|
"""
|
|
|
|
course = courses.get_course_by_id(course_id)
|
|
|
|
if not course.is_cohorted:
|
|
# this is the easy case :)
|
|
ans = []
|
|
else:
|
|
ans = course.cohorted_discussions
|
|
|
|
return ans
|
|
|
|
|
|
def get_cohort(user, course_id):
|
|
"""
|
|
Given a django User and a course_id, return the user's cohort in that
|
|
cohort.
|
|
|
|
Arguments:
|
|
user: a Django User object.
|
|
course_id: string in the format 'org/course/run'
|
|
|
|
Returns:
|
|
A CourseUserGroup object if the course is cohorted and the User has a
|
|
cohort, else None.
|
|
|
|
Raises:
|
|
ValueError if the course_id doesn't exist.
|
|
"""
|
|
# First check whether the course is cohorted (users shouldn't be in a cohort
|
|
# in non-cohorted courses, but settings can change after course starts)
|
|
try:
|
|
course = courses.get_course_by_id(course_id)
|
|
except Http404:
|
|
raise ValueError("Invalid course_id")
|
|
|
|
if not course.is_cohorted:
|
|
return None
|
|
|
|
try:
|
|
return CourseUserGroup.objects.get(course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT,
|
|
users__id=user.id)
|
|
except CourseUserGroup.DoesNotExist:
|
|
# Didn't find the group. We'll go on to create one if needed.
|
|
pass
|
|
|
|
if not course.auto_cohort:
|
|
return None
|
|
|
|
choices = course.auto_cohort_groups
|
|
n = len(choices)
|
|
if n == 0:
|
|
# Nowhere to put user
|
|
log.warning("Course %s is auto-cohorted, but there are no"
|
|
" auto_cohort_groups specified",
|
|
course_id)
|
|
return None
|
|
|
|
# Put user in a random group, creating it if needed
|
|
group_name = local_random().choice(choices)
|
|
|
|
group, created = CourseUserGroup.objects.get_or_create(
|
|
course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT,
|
|
name=group_name)
|
|
|
|
user.course_groups.add(group)
|
|
return group
|
|
|
|
|
|
def get_course_cohorts(course_id):
|
|
"""
|
|
Get a list of all the cohorts in the given course.
|
|
|
|
Arguments:
|
|
course_id: string in the format 'org/course/run'
|
|
|
|
Returns:
|
|
A list of CourseUserGroup objects. Empty if there are no cohorts. Does
|
|
not check whether the course is cohorted.
|
|
"""
|
|
return list(CourseUserGroup.objects.filter(course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT))
|
|
|
|
### Helpers for cohort management views
|
|
|
|
|
|
def get_cohort_by_name(course_id, name):
|
|
"""
|
|
Return the CourseUserGroup object for the given cohort. Raises DoesNotExist
|
|
it isn't present.
|
|
"""
|
|
return CourseUserGroup.objects.get(course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT,
|
|
name=name)
|
|
|
|
|
|
def get_cohort_by_id(course_id, cohort_id):
|
|
"""
|
|
Return the CourseUserGroup object for the given cohort. Raises DoesNotExist
|
|
it isn't present. Uses the course_id for extra validation...
|
|
"""
|
|
return CourseUserGroup.objects.get(course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT,
|
|
id=cohort_id)
|
|
|
|
|
|
def add_cohort(course_id, name):
|
|
"""
|
|
Add a cohort to a course. Raises ValueError if a cohort of the same name already
|
|
exists.
|
|
"""
|
|
log.debug("Adding cohort %s to %s", name, course_id)
|
|
if CourseUserGroup.objects.filter(course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT,
|
|
name=name).exists():
|
|
raise ValueError("Can't create two cohorts with the same name")
|
|
|
|
return CourseUserGroup.objects.create(course_id=course_id,
|
|
group_type=CourseUserGroup.COHORT,
|
|
name=name)
|
|
|
|
|
|
class CohortConflict(Exception):
|
|
"""
|
|
Raised when user to be added is already in another cohort in same course.
|
|
"""
|
|
pass
|
|
|
|
|
|
def add_user_to_cohort(cohort, username_or_email):
|
|
"""
|
|
Look up the given user, and if successful, add them to the specified cohort.
|
|
|
|
Arguments:
|
|
cohort: CourseUserGroup
|
|
username_or_email: string. Treated as email if has '@'
|
|
|
|
Returns:
|
|
User object.
|
|
|
|
Raises:
|
|
User.DoesNotExist if can't find user.
|
|
|
|
ValueError if user already present in this cohort.
|
|
|
|
CohortConflict if user already in another cohort.
|
|
"""
|
|
user = get_user_by_username_or_email(username_or_email)
|
|
|
|
# If user in any cohorts in this course already, complain
|
|
course_cohorts = CourseUserGroup.objects.filter(
|
|
course_id=cohort.course_id,
|
|
users__id=user.id,
|
|
group_type=CourseUserGroup.COHORT)
|
|
if course_cohorts.exists():
|
|
if course_cohorts[0] == cohort:
|
|
raise ValueError("User {0} already present in cohort {1}".format(
|
|
user.username,
|
|
cohort.name))
|
|
else:
|
|
raise CohortConflict("User {0} is in another cohort {1} in course"
|
|
.format(user.username,
|
|
course_cohorts[0].name))
|
|
|
|
cohort.users.add(user)
|
|
return user
|
|
|
|
|
|
def get_course_cohort_names(course_id):
|
|
"""
|
|
Return a list of the cohort names in a course.
|
|
"""
|
|
return [c.name for c in get_course_cohorts(course_id)]
|
|
|
|
|
|
def delete_empty_cohort(course_id, name):
|
|
"""
|
|
Remove an empty cohort. Raise ValueError if cohort is not empty.
|
|
"""
|
|
cohort = get_cohort_by_name(course_id, name)
|
|
if cohort.users.exists():
|
|
raise ValueError(
|
|
"Can't delete non-empty cohort {0} in course {1}".format(
|
|
name, course_id))
|
|
|
|
cohort.delete()
|