""" 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)