feat: UpstreamSyncMixin

cms/envs/common.py

@@ -1020,6 +1020,7 @@
     XModuleMixin,
     EditInfoMixin,
     AuthoringMixin,
+    "openedx.core.djangoapps.content_libraries.sync.UpstreamSyncMixin",
 )
 
 # .. setting_name: XBLOCK_EXTRA_MIXINS

openedx/core/djangoapps/content_libraries/sync.py

@@ -0,0 +1,229 @@
+"""
+Synchronize content and settings from upstream blocks (in content libraries) to their
+downstream usages (in courses, etc.)
+
+At the time of writing, upstream blocks are assumed to come from content libraries.
+However, the XBlock fields are designed to be agnostic to their upstream's source context,
+so this assumption could be relaxed in the future if there is a need for upstreams from
+other kinds of learning contexts.
+"""
+import json
+
+from django.contrib.auth import get_user_model
+from opaque_keys import InvalidKeyError
+from opaque_keys.edx.keys import UsageKey
+from opaque_keys.edx.locator import LibraryUsageLocatorV2
+from xblock.fields import Scope, String, Integer, List, Dict
+from xblock.core import XBlockMixin, XBlock
+from webob import Request, Response
+
+from openedx.core.djangoapps.content_libraries.api import (
+    get_library_block,
+    LibraryXBlockMetadata,
+    ContentLibraryBlockNotFound,
+)
+from openedx.core.djangoapps.xblock.api import load_block, NotFound as XBlockNotFound
+
+
+class UpstreamSyncMixin(XBlockMixin):
+    """
+    @@TODO docstring
+    """
+
+    upstream = String(
+        scope=Scope.settings,
+        help=(
+            "The usage key of a block (generally within a Content Library) which serves as a source of upstream "
+            "updates for this block, or None if there is no such upstream. Please note: It is valid for upstream_block "
+            "to hold a usage key for a block that does not exist (or does not *yet* exist) on this instance, "
+            "particularly if this block was imported from a different instance."
+        ),
+        hidden=True,
+        default=None,
+        enforce_type=True,
+    )
+    upstream_version = Integer(
+        scope=Scope.settings,
+        help=(
+            "The upstream_block's version number, at the time this block was created from it. "
+            "If this version is older than the upstream_block's latest version, then CMS will "
+            "allow this block to fetch updated content from upstream_block."
+        ),
+        hidden=True,
+        default=None,
+        enforce_type=True,
+    )
+    upstream_overridden = List(
+        scope=Scope.settings,
+        help=(
+            "@@TODO helptext"
+        ),
+        hidden=True,
+        default=[],
+        enforce_type=True,
+    )
+    upstream_settings = Dict(
+        scope=Scope.settings,
+        help=(
+            "@@TODO helptext"
+        ),
+        hidden=True,
+        default={},
+        enforce_type=True,
+    )
+
+    def save(self, *args, **kwargs):
+        """
+        @@TODO docstring
+        @@TODO use is_dirty instead of getattr for efficiency?
+        """
+        for field_name, value in self.upstream_settings.items():
+            if field_name not in self.upstream_overridden:
+                if value != getattr(self, field_name):
+                    self.upstream_overridden.append(field_name)
+        super().save()
+
+    def assign_upstream(self, upstream_key: LibraryUsageLocatorV2 | None) -> LibraryXBlockMetadata | None:
+        """
+        Assign an upstream to this block and fetch upstream settings.
+
+        Does not save block; caller must do so.
+
+        Raises: ContentLibraryBlockNotFound, NotFound
+        """
+        old_upstream = self.upstream
+        self.upstream = str(upstream_key)
+        try:
+            return self._sync_with_upstream(apply_updates=False)
+        except (ContentLibraryBlockNotFound, XBlockNotFound):
+            self.upstream = old_upstream
+            raise
+
+    @XBlock.handler
+    def upstream_link(self, request: Request, _suffix=None) -> Response:
+        """
+        @@TODO docstring
+
+        GET: Retrieve upstream link
+        PUT: Set upstream link
+        DELETE: Remove upstream link
+
+        200: Success, with JSON data on upstream link
+        204: Success, no upstream link
+        400: Bad request data
+        401: Unauthenticated
+        405: Bad method
+        """
+        if request.method == "DELETE":
+            self.assign_upstream(None)
+            self.save()
+            return Response(status_code=204)
+        if request.method in ["PUT", "GET"]:
+            if request.method == "PUT":
+                try:
+                    usage_key_string = json.loads(request.data["usage_key"])
+                except json.JSONDecodeError:
+                    return Response("bad json", status_code=400)
+                except KeyError:
+                    return Response("missing top-level key in json body: usage_key", status_code=400)
+                try:
+                    usage_key = LibraryUsageLocatorV2.from_string(usage_key_string)
+                except InvalidKeyError:
+                    return Response(f"invalid library block key: {usage_key_string}", status_code=400)
+                try:
+                    upstream_meta = self.assign_upstream(usage_key)  # type: ignore[assignment]
+                except ContentLibraryBlockNotFound:
+                    return Response("could not load library block metadata: {usage key}", status_code=400)
+                self.save()
+            if request.method == "GET":
+                try:
+                    upstream_meta = self.get_upstream_meta()
+                except InvalidKeyError:
+                    return Response("upstream is not a valid usage key: {self.upstream}", status_code=400)
+                except ContentLibraryBlockNotFound:
+                    return Response("could not load upstream block metadata: {usage key}", status_code=400)
+                if not upstream_meta:
+                    return Response(status_code=204)
+            return Response(
+                json.dumps(
+                    {
+                        "usage_key": self.upstream,
+                        "version_current": self.upstream_version,
+                        "version_latest": upstream_meta.version_num if upstream_meta else None,
+                    },
+                    indent=4,
+                ),
+            )
+        return Response(status_code=405)
+
+    @XBlock.handler
+    def update_from_upstream(self, request: Request, suffix=None) -> Response:
+        """
+        @@TODO docstring
+        """
+        if request.method != "POST":
+            return Response(status_code=405)
+        try:
+            self._sync_with_upstream(apply_updates=True)
+        except BadUpstream as exc:
+            return Response(str(exc), status_code=400)
+        self.save()
+        return Response(status_code=204)
+
+    def _sync_with_upstream(self, *, apply_updates: bool) -> LibraryXBlockMetadata | None:
+        """
+        @@TODO docstring
+
+        Does not save block; caller must do so.
+
+        Raises: InvalidKeyError, ContentLibraryBlockNotFound, XBlockNotFound
+        """
+        upstream_meta = self.get_upstream_meta()
+        if not upstream_meta:
+            self.upstream_overridden = []
+            self.upstream_version = None
+            return None
+        self.upstream_settings = {}
+        # @@TODO: do we need user_id to get the block? if so, is there a better way to get it?
+        user_id = self.runtime.service(self, "user")._django_user.id  # pylint: disable=protected-access
+        upstream_block = load_block(upstream_meta.usage_key, get_user_model().objects.get(id=user_id))
+        for field_name, field in upstream_block.fields.items():
+            if field.scope not in [Scope.settings, Scope.content]:
+                continue
+            value = getattr(upstream_block, field_name)
+            if field.scope == Scope.settings:
+                self.upstream_settings[field_name] = value
+                if field_name in self.upstream_overridden:
+                    continue
+            if not apply_updates:
+                continue
+            setattr(self, field_name, value)
+        self.upstream_version = upstream_meta.version_num
+        return upstream_meta
+
+    def get_upstream_meta(self) -> LibraryXBlockMetadata | None:
+        """
+        @@TODO docstring
+
+        Raises: InvalidKeyError, ContentLibraryBlockNotFound
+        """
+        if not self.upstream:
+            return None
+        upstream_key = LibraryUsageLocatorV2.from_string(self.upstream)
+        return get_library_block(upstream_key)
+
+
+class BadUpstream(Exception):
+    """
+    Base exception for any content-level problems we can hit while loading a block's upstream.
+
+    Should not represent unexpected internal server errors.
+    May appear in API responses, so they should be somewhat user friendly and avoid sensitive info.
+    """
+
+
+def is_valid_upstream(usage_key: UsageKey) -> bool:
+    """
+    @@TODO docstring
+    """
+    return isinstance(usage_key, LibraryUsageLocatorV2)