"""
Helper functions for using the models, so external
apps don't tie functionality to internal implementation.
"""
from __future__ import unicode_literals
import logging
from os import walk, sep
from os.path import join
from django.contrib.auth.models import User
from django.core.files import File
from django.db import transaction
from guardian.shortcuts import get_objects_for_user, get_perms
from learningresources.models import (
Course, Repository, LearningResource, LearningResourceType, StaticAsset
)
from roles.permissions import RepoPermission
log = logging.getLogger(__name__)
[docs]class LearningResourceException(Exception):
"""Base class for our custom exceptions."""
pass
[docs]class PermissionDenied(LearningResourceException):
"""
Raised by the API when the requested item exists, but the
user is not allowed to access it.
"""
pass
[docs]class NotFound(LearningResourceException):
"""Raised by the API when the item requested does not exist."""
pass
[docs]class MissingTitle(object):
"""
Class to describe the missing title for importer and for
the description path
"""
for_title_field = 'Missing Title'
for_desc_path_field = '...'
[docs]def create_course(org, repo_id, course_number, run, user_id):
"""
Add a course to the database.
Args:
org (unicode): Organization
repo_id (int): Repository id
course_number (unicode): Course number
run (unicode): Run
user_id (int): Primary key of user creating the course
Raises:
ValueError: Duplicate course
Returns:
course (learningresources.models.Course): The created course
"""
# Check on unique values before attempting a get_or_create, because
# items such as date_created will always make it non-unique.
unique = {
"org": org, "course_number": course_number, "run": run,
"repository_id": repo_id,
}
if Course.objects.filter(**unique).exists():
raise ValueError("Duplicate course")
kwargs = {
"org": org,
"course_number": course_number,
"run": run,
'imported_by_id': user_id,
"repository_id": repo_id,
}
with transaction.atomic():
course = Course.objects.create(**kwargs)
return course
# pylint: disable=too-many-arguments
[docs]def create_resource(
course, parent, resource_type, title, content_xml, mpath, url_name,
dpath
):
"""
Create a learning resource.
Args:
course (learningresources.models.Course): Course
parent (learningresources.models.LearningResource):
Parent LearningResource
resource_type (unicode): Name of LearningResourceType
title (unicode): Title of resource
content_xml (unicode): XML
mpath (unicode): Materialized path
url_name (unicode): Resource identifier
dpath (unicode): Description path
Returns:
resource (learningresources.models.LearningResource):
New LearningResource
"""
params = {
"course": course,
"learning_resource_type_id": type_id_by_name(resource_type),
"title": title,
"content_xml": content_xml,
"materialized_path": mpath,
"url_name": url_name,
"description_path": dpath,
}
if parent is not None:
params["parent_id"] = parent.id
with transaction.atomic():
return LearningResource.objects.create(**params)
[docs]def type_id_by_name(name):
"""
Get or create a LearningResourceType by name.
This would do fewer queries if it did all the lookups up front, but
this is simpler to read and understand and still prevents most lookups.
Also, it can't prevent inserts, so it's never guaranteed to be just
a single query.
Args:
name (unicode): LearningResourceType.name
Returns:
type_id (int): Primary key of learningresources.LearningResourceType
"""
with transaction.atomic():
obj, _ = LearningResourceType.objects.get_or_create(name=name.lower())
return obj.id
[docs]def get_repos(user_id):
"""
Get all repositories a user may access.
Args:
user_id (int): Primary key of user
Returns:
repos (query set of learningresource.Repository): Repositories
"""
try:
user = User.objects.get(id=user_id)
except User.DoesNotExist:
raise PermissionDenied(
"user does not have permission for this repository"
)
return get_objects_for_user(
user,
RepoPermission.view_repo[0],
klass=Repository
)
[docs]def get_repo(repo_slug, user_id):
"""
Get repository for a user if s/he can access it.
Returns a repository object if it exists or
* raises a 404 if the object does not exist
* raises a 403 if the object exists but the user doesn't have permission
Args:
repo_slug (unicode): Repository slug
user_id (int): Primary key of user
Returns:
repo (learningresource.Repository): Repository
"""
try:
user = User.objects.get(id=user_id)
except User.DoesNotExist:
raise PermissionDenied(
"user does not have permission for this repository")
try:
repo = Repository.objects.get(slug=repo_slug)
except Repository.DoesNotExist:
raise NotFound()
if RepoPermission.view_repo[0] in get_perms(user, repo):
return repo
raise PermissionDenied("user does not have permission for this repository")
[docs]def create_repo(name, description, user_id):
"""
Create a new repository.
Args:
name (unicode): Repository name
description (unicode): Repository description
user_id (int): User ID of repository creator
Returns:
repo (learningresources.Repository): Newly-created repository
"""
with transaction.atomic():
return Repository.objects.create(
name=name, description=description,
created_by_id=user_id,
)
[docs]def get_resources(repo_id):
"""
Get resources from a repository ordered by title.
Args:
repo_id (int): Primary key of the repository
Returns:
list (list of learningresources.LearningResource): List of resources
"""
return LearningResource.objects.select_related(
"learning_resource_type", "course__repository").filter(
course__repository__id=repo_id).order_by("title")
[docs]def get_resource(resource_id, user_id):
"""
Get single resource.
Args:
resource_id (int): Primary key of the LearningResource
user_id (int): Primary key of the user requesting the resource
Returns:
resource (learningresources.LearningResource): Resource
May be None if the resource does not exist or the user does
not have permissions.
"""
try:
resource = LearningResource.objects.get(id=resource_id)
except LearningResource.DoesNotExist:
raise NotFound("LearningResource not found")
# This will raise PermissionDenied if it fails.
get_repo(resource.course.repository.slug, user_id)
return resource
[docs]def create_static_asset(course_id, handle):
"""
Create a static asset.
Args:
course_id (int): learningresources.models.Course pk
handle (django.core.files.File): file handle
Returns:
learningresources.models.StaticAsset
"""
with transaction.atomic():
return StaticAsset.objects.create(course_id=course_id, asset=handle)
def _subs_filename(subs_id, lang='en'):
"""
Generate proper filename for storage.
Function copied from:
edx-platform/common/lib/xmodule/xmodule/video_module/transcripts_utils.py
Args:
subs_id (str): Subs id string
lang (str): Locale language (optional) default: en
Returns:
filename (str): Filename of subs file
"""
if lang in ('en', "", None):
return u'subs_{0}.srt.sjson'.format(subs_id)
else:
return u'{0}_subs_{1}.srt.sjson'.format(lang, subs_id)
[docs]def get_video_sub(xml):
"""
Get subtitle IDs from <video> XML.
Args:
xml (lxml.etree): xml for a LearningResource
Returns:
sub string: subtitle string
"""
subs = xml.xpath("@sub")
# It's not possible to have more than one.
if len(subs) == 0:
return ""
return _subs_filename(subs[0])
[docs]def import_static_assets(course, path):
"""
Upload all assets and create model records of them for a given
course and path.
Args:
course (learningresources.models.Course): Course to add assets to.
path (unicode): course specific path to extracted OLX tree.
Returns:
None
"""
for root, _, files in walk(path):
for name in files:
with open(join(root, name), 'rb') as open_file:
django_file = File(open_file)
# Remove base path from file name
name = join(root, name).replace(path + sep, '', 1)
django_file.name = name
create_static_asset(course.id, django_file)
[docs]def update_xanalytics(data):
"""
Update xanalytics fields for a LearningResource.
Args:
data (dict): dict from JSON file from xanalytics
Returns:
count (int): number of records updated
"""
vals = data.get("module_medata", [])
course_number = data.get("course_id", "")
count = 0
for rec in vals:
resource_key = rec.pop("module_id")
count = LearningResource.objects.filter(
uuid=resource_key,
course__course_number=course_number,
).update(**rec)
if count is None:
count = 0
return count
[docs]def join_description_paths(*args):
"""
Helper function to format the description path.
Args:
args (unicode): description path
Returns:
unicode: Formatted dpath
"""
return ' / '.join([dpath for dpath in args if dpath != ''])
[docs]def update_description_path(resource, force_parent_update=False):
"""
Updates the specified learning resource description path
based on the current title and the parent's description path
Args:
resource (learningresources.models.LearningResource): LearningResource
force_parent_update (boolean): force parent update
Returns:
None
"""
description_path = ''
title_desc_path = resource.title
if title_desc_path == MissingTitle.for_title_field:
title_desc_path = MissingTitle.for_desc_path_field
if resource.parent is None:
description_path = join_description_paths(title_desc_path)
else:
# if the parent doesn't have a description_path update first the parent
if resource.parent.description_path == '' or force_parent_update:
update_description_path(resource.parent, force_parent_update)
# the current description path is
# the parent's one plus the current title
description_path = join_description_paths(
resource.parent.description_path,
title_desc_path
)
resource.description_path = description_path
resource.save()