92b684004e
This re-applies commit 36cc415 with handling an invalid context_key in the
`PartitionService`. It can happen when rendering a `LibraryContentBlock` in
Studio because this service is initialized by the modulestore when validating
an XBlock to gather its error messages in the `studio_xblock_wrapper`.
311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""
|
|
Module contains various XModule/XBlock services
|
|
"""
|
|
|
|
|
|
import inspect
|
|
import logging
|
|
from functools import partial
|
|
|
|
from config_models.models import ConfigurationModel
|
|
from django.conf import settings
|
|
from eventtracking import tracker
|
|
from edx_when.field_data import DateLookupFieldData
|
|
from xblock.reference.plugins import Service
|
|
from xblock.runtime import KvsFieldData
|
|
|
|
from common.djangoapps.track import contexts
|
|
from lms.djangoapps.courseware.masquerade import is_masquerading_as_specific_student
|
|
from xmodule.modulestore.django import modulestore
|
|
|
|
from lms.djangoapps.courseware.field_overrides import OverrideFieldData
|
|
from lms.djangoapps.courseware.model_data import DjangoKeyValueStore, FieldDataCache
|
|
from lms.djangoapps.lms_xblock.field_data import LmsFieldData
|
|
from lms.djangoapps.lms_xblock.models import XBlockAsidesConfig
|
|
|
|
from lms.djangoapps.grades.api import signals as grades_signals
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class SettingsService:
|
|
"""
|
|
Allows server-wide configuration of XBlocks on a per-type basis
|
|
|
|
XBlock settings are read from XBLOCK_SETTINGS settings key. Each XBlock is allowed access
|
|
to single settings bucket. Bucket is determined by this service using the following rules:
|
|
|
|
* Value of SettingsService.xblock_settings_bucket_selector is examined. If XBlock have attribute/property
|
|
with the name of that value this attribute/property is read to get the bucket key (e.g. if XBlock have
|
|
`block_settings_key = 'my_block_settings'`, bucket key would be 'my_block_settings').
|
|
* Otherwise, XBlock class name is used
|
|
|
|
Service is content-agnostic: it just returns whatever happen to be in the settings bucket (technically, it returns
|
|
the bucket itself).
|
|
|
|
If `default` argument is specified it is returned if:
|
|
* There are no XBLOCK_SETTINGS setting
|
|
* XBLOCK_SETTINGS is empty
|
|
* XBLOCK_SETTINGS does not contain settings bucket
|
|
|
|
If `default` is not specified or None, empty dictionary is used for default.
|
|
|
|
Example:
|
|
|
|
"XBLOCK_SETTINGS": {
|
|
"my_block": {
|
|
"setting1": 1,
|
|
"setting2": []
|
|
},
|
|
"my_other_block": [1, 2, 3],
|
|
"MyThirdBlock": "QWERTY"
|
|
}
|
|
|
|
class MyBlock: block_settings_key='my_block'
|
|
class MyOtherBlock: block_settings_key='my_other_block'
|
|
class MyThirdBlock: pass
|
|
class MissingBlock: pass
|
|
|
|
service = SettingsService()
|
|
service.get_settings_bucket(MyBlock()) # { "setting1": 1, "setting2": [] }
|
|
service.get_settings_bucket(MyOtherBlock()) # [1, 2, 3]
|
|
service.get_settings_bucket(MyThirdBlock()) # "QWERTY"
|
|
service.get_settings_bucket(MissingBlock()) # {}
|
|
service.get_settings_bucket(MissingBlock(), "default") # "default"
|
|
service.get_settings_bucket(MissingBlock(), None) # {}
|
|
"""
|
|
xblock_settings_bucket_selector = 'block_settings_key'
|
|
|
|
def get_settings_bucket(self, block, default=None):
|
|
""" Gets xblock settings dictionary from settings. """
|
|
if not block:
|
|
raise ValueError(f"Expected XBlock instance, got {block} of type {type(block)}")
|
|
|
|
actual_default = default if default is not None else {}
|
|
xblock_settings_bucket = getattr(block, self.xblock_settings_bucket_selector, block.unmixed_class.__name__)
|
|
xblock_settings = settings.XBLOCK_SETTINGS if hasattr(settings, "XBLOCK_SETTINGS") else {}
|
|
return xblock_settings.get(xblock_settings_bucket, actual_default)
|
|
|
|
|
|
# TODO: ConfigurationService and its usage will be removed as a part of EDUCATOR-121
|
|
# reference: https://openedx.atlassian.net/browse/EDUCATOR-121
|
|
class ConfigurationService:
|
|
"""
|
|
An XBlock service to talk with the Configuration Models. This service should provide
|
|
a pathway to Configuration Model which is designed to configure the corresponding XBlock.
|
|
"""
|
|
def __init__(self, configuration_model):
|
|
"""
|
|
Class initializer, this exposes configuration model to XBlock.
|
|
|
|
Arguments:
|
|
configuration_model (ConfigurationModel): configurations for an XBlock
|
|
|
|
Raises:
|
|
exception (ValueError): when configuration_model is not a subclass of
|
|
ConfigurationModel.
|
|
"""
|
|
if not (inspect.isclass(configuration_model) and issubclass(configuration_model, ConfigurationModel)):
|
|
raise ValueError(
|
|
"Expected ConfigurationModel got {} of type {}".format(
|
|
configuration_model,
|
|
type(configuration_model)
|
|
)
|
|
)
|
|
|
|
self.configuration = configuration_model
|
|
|
|
|
|
class TeamsConfigurationService:
|
|
"""
|
|
An XBlock service that returns the teams_configuration object for a course.
|
|
"""
|
|
def __init__(self):
|
|
self._course = None
|
|
|
|
def get_course(self, course_id):
|
|
"""
|
|
Return the course instance associated with this TeamsConfigurationService.
|
|
This default implementation looks up the course from the modulestore.
|
|
"""
|
|
return modulestore().get_course(course_id)
|
|
|
|
def get_teams_configuration(self, course_id):
|
|
"""
|
|
Returns the team configuration for a given course.id
|
|
"""
|
|
if not self._course:
|
|
self._course = self.get_course(course_id)
|
|
return self._course.teams_configuration
|
|
|
|
|
|
class RebindUserServiceError(Exception):
|
|
pass
|
|
|
|
|
|
class RebindUserService(Service):
|
|
"""
|
|
An XBlock Service that allows modules to get rebound to real users if it was previously bound to an AnonymousUser.
|
|
|
|
This used to be a local function inside the `lms.djangoapps.courseware.block_render.prepare_runtime_for_user`
|
|
method. This has been refactored out into a service and lives in this module temporarily.
|
|
|
|
TODO: Only the old LTI XBlock uses it in 2 places for LTI 2.0 integration. As the LTI XBlock is deprecated in
|
|
favour of the LTI Consumer XBlock, this should be removed when the LTI XBlock is removed.
|
|
|
|
Arguments:
|
|
user (User) - A Django User object
|
|
course_id (str) - Course ID
|
|
course (Course) - Course Object
|
|
kwargs (dict) - all the keyword arguments that need to be passed to the `prepare_runtime_for_user`
|
|
function when it is called during rebinding
|
|
"""
|
|
def __init__(self, user, course_id, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.user = user
|
|
self.course_id = course_id
|
|
self._kwargs = kwargs
|
|
|
|
def rebind_noauth_module_to_user(self, block, real_user):
|
|
"""
|
|
Function that rebinds the module to the real_user.
|
|
|
|
Will only work within a module bound to an AnonymousUser, e.g. one that's instantiated by the noauth_handler.
|
|
|
|
Arguments:
|
|
block (any xblock type): the module to rebind
|
|
real_user (django.contrib.auth.models.User): the user to bind to
|
|
|
|
Returns:
|
|
nothing (but the side effect is that module is re-bound to real_user)
|
|
"""
|
|
if self.user.is_authenticated:
|
|
err_msg = "rebind_noauth_module_to_user can only be called from a module bound to an anonymous user"
|
|
log.error(err_msg)
|
|
raise RebindUserServiceError(err_msg)
|
|
|
|
field_data_cache_real_user = FieldDataCache.cache_for_block_descendents(
|
|
self.course_id,
|
|
real_user,
|
|
block,
|
|
asides=XBlockAsidesConfig.possible_asides(),
|
|
)
|
|
student_data_real_user = KvsFieldData(DjangoKeyValueStore(field_data_cache_real_user))
|
|
|
|
with modulestore().bulk_operations(self.course_id):
|
|
course = modulestore().get_course(course_key=self.course_id)
|
|
|
|
from lms.djangoapps.courseware.block_render import prepare_runtime_for_user
|
|
prepare_runtime_for_user(
|
|
user=real_user,
|
|
student_data=student_data_real_user, # These have implicit user bindings, rest of args considered not to
|
|
runtime=block.runtime,
|
|
course_id=self.course_id,
|
|
course=course,
|
|
**self._kwargs
|
|
)
|
|
|
|
block.bind_for_student(
|
|
real_user.id,
|
|
[
|
|
partial(DateLookupFieldData, course_id=self.course_id, user=self.user),
|
|
partial(OverrideFieldData.wrap, real_user, course),
|
|
partial(LmsFieldData, student_data=student_data_real_user),
|
|
],
|
|
)
|
|
|
|
|
|
class EventPublishingService(Service):
|
|
"""
|
|
An XBlock Service that allows XModules to publish events (e.g. grading, completion).
|
|
|
|
We have implemented it as a seperate service to be able to alter its behavior when using
|
|
a different context: LMS, Studio, or Instructor tasks.
|
|
"""
|
|
def __init__(self, user, course_id, track_function, **kwargs):
|
|
super().__init__(**kwargs)
|
|
self.user = user
|
|
self.course_id = course_id
|
|
self.track_function = track_function
|
|
self.completion_service = None
|
|
|
|
def publish(self, block, event_type, event):
|
|
"""
|
|
A function that allows XModules to publish events.
|
|
"""
|
|
self.completion_service = block.runtime.service(block, 'completion')
|
|
|
|
handle_event = self._get_event_handler(event_type)
|
|
if handle_event and not is_masquerading_as_specific_student(self.user, self.course_id):
|
|
handle_event(block, event)
|
|
else:
|
|
context = contexts.course_context_from_course_id(self.course_id)
|
|
if not self.user.is_anonymous:
|
|
context['user_id'] = self.user.id
|
|
|
|
context['asides'] = {}
|
|
for aside in block.runtime.get_asides(block):
|
|
if hasattr(aside, 'get_event_context'):
|
|
aside_event_info = aside.get_event_context(event_type, event)
|
|
if aside_event_info is not None:
|
|
context['asides'][aside.scope_ids.block_type] = aside_event_info
|
|
with tracker.get_tracker().context(event_type, context):
|
|
self.track_function(event_type, event)
|
|
|
|
def _get_event_handler(self, event_type):
|
|
"""
|
|
Return an appropriate function to handle the event.
|
|
|
|
Returns None if no special processing is required.
|
|
"""
|
|
handlers = {
|
|
'grade': self._handle_grade_event,
|
|
}
|
|
if self.completion_service and self.completion_service.completion_tracking_enabled():
|
|
handlers.update(
|
|
{
|
|
'completion': lambda block, event: self.completion_service.submit_completion(
|
|
block.scope_ids.usage_id, event['completion']
|
|
),
|
|
'progress': self._handle_deprecated_progress_event,
|
|
}
|
|
)
|
|
return handlers.get(event_type)
|
|
|
|
def _handle_grade_event(self, block, event):
|
|
"""
|
|
Submit a grade for the block.
|
|
"""
|
|
if not self.user.is_anonymous:
|
|
grades_signals.SCORE_PUBLISHED.send(
|
|
sender=None,
|
|
block=block,
|
|
user=self.user,
|
|
raw_earned=event['value'],
|
|
raw_possible=event['max_value'],
|
|
only_if_higher=event.get('only_if_higher'),
|
|
score_deleted=event.get('score_deleted'),
|
|
grader_response=event.get('grader_response'),
|
|
)
|
|
|
|
def _handle_deprecated_progress_event(self, block, event):
|
|
"""
|
|
DEPRECATED: Submit a completion for the block represented by the
|
|
progress event.
|
|
|
|
This exists to support the legacy progress extension used by
|
|
edx-solutions. New XBlocks should not emit these events, but instead
|
|
emit completion events directly.
|
|
"""
|
|
requested_user_id = event.get('user_id', self.user.id)
|
|
if requested_user_id != self.user.id:
|
|
log.warning(f"{self.user} tried to submit a completion on behalf of {requested_user_id}")
|
|
return
|
|
|
|
# If blocks explicitly declare support for the new completion API,
|
|
# we expect them to emit 'completion' events,
|
|
# and we ignore the deprecated 'progress' events
|
|
# in order to avoid duplicate work and possibly conflicting semantics.
|
|
if not getattr(block, 'has_custom_completion', False):
|
|
self.completion_service.submit_completion(block.scope_ids.usage_id, 1.0)
|