openedx.core.djangoapps.content_libraries package

Contents

openedx.core.djangoapps.content_libraries package#

Subpackages#

Submodules#

openedx.core.djangoapps.content_libraries.apps module#

Django AppConfig for Content Libraries Implementation

class openedx.core.djangoapps.content_libraries.apps.ContentLibrariesConfig(app_name, app_module)#

Bases: AppConfig

Django AppConfig for Content Libraries Implementation

name = 'openedx.core.djangoapps.content_libraries'#
plugin_app = {'settings_config': {'cms.djangoapp': {}}, 'url_config': {'cms.djangoapp': {'namespace': 'content_libraries'}}}#
ready()#

Import signal handler’s module to ensure they are registered.

verbose_name = 'Content Libraries'#

openedx.core.djangoapps.content_libraries.constants module#

Constants used for the content libraries.

openedx.core.djangoapps.content_libraries.library_context module#

Definition of “Library” as a learning context.

class openedx.core.djangoapps.content_libraries.library_context.LibraryContextImpl(**kwargs)#

Bases: LearningContext

Implements content libraries as a learning context.

This is the new content libraries based on openedx_content, not the old content libraries based on modulestore.

block_exists(usage_key: LibraryUsageLocatorV2)#

Does the block for this usage_key exist in this Library?

Note that this applies to all versions, i.e. you can put a usage key for a piece of content that has been soft-deleted (removed from Drafts), and it will still return True here. That’s because for the purposes of permission checking, we just want to know whether that block has ever existed in this Library, because we could be looking at any older version of it.

can_edit_block(user: User, usage_key: UsageKeyV2) → bool#

Assuming a block with the specified ID (usage_key) exists, does the specified user have permission to edit it (make changes to the fields / authored data store)?

May raise ContentLibraryNotFound if the library does not exist.

can_view_block(user: User, usage_key: UsageKeyV2) → bool#

Does the specified usage key exist in its context, and if so, does the specified user have permission to view it and interact with it (call handlers, save user state, etc.)?

May raise ContentLibraryNotFound if the library does not exist.

can_view_block_for_editing(user: User, usage_key: UsageKeyV2) → bool#

Assuming a block with the specified ID (usage_key) exists, does the specified user have permission to view its fields and OLX details (but not necessarily to make changes to it)?

May raise ContentLibraryNotFound if the library does not exist.

openedx.core.djangoapps.content_libraries.models module#

Content Libraries Models#

This module contains the models for new Content Libraries.

class openedx.core.djangoapps.content_libraries.models.ContentLibrary(*args, **kwargs)#

Bases: Model

A Content Library is a collection of content (XBlocks and/or static assets)

All actual content is stored in openedx_content, and any data that we’d want to transfer to another instance if this library were exported and then re-imported on another Open edX instance should be kept in openedx_content. This model in Studio should only be used to track settings specific to this Open edX instance, like who has permission to edit this content library.

exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

allow_public_learning#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

allow_public_read#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

authz_scopes#

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

get_license_display(*, field=<django.db.models.fields.CharField: license>)#
id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

learning_package#

Accessor to the related object on the forward side of a one-to-one relation.

In the example:

class Restaurant(Model):
    place = OneToOneField(Place, related_name='restaurant')

Restaurant.place is a ForwardOneToOneDescriptor instance.

learning_package_id#
property library_key#

Get the LibraryLocatorV2 opaque key for this library

license#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects: ClassVar[ContentLibraryManager] = <openedx.core.djangoapps.content_libraries.models.ContentLibraryManager object>#
org#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

org_id#
permission_grants#

Accessor to the related objects manager on the reverse side of a many-to-one relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Parent.children is a ReverseManyToOneDescriptor instance.

Most of the implementation is delegated to a dynamically defined manager class built by create_forward_many_to_many_manager() defined below.

slug#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

class openedx.core.djangoapps.content_libraries.models.ContentLibraryManager(*args, **kwargs)#

Bases: Manager

Custom manager for ContentLibrary class.

get_by_key(library_key) → ContentLibrary#

Get the ContentLibrary for the given LibraryLocatorV2 key.

class openedx.core.djangoapps.content_libraries.models.ContentLibraryPermission(*args, **kwargs)#

Bases: Model

Row recording permissions for a content library

Deprecated openedx/openedx-platform#37409.

ACCESS_LEVEL_CHOICES = (('admin', 'Administer users and author content'), ('author', 'Author content'), ('read', 'Read-only'))#
ADMIN_LEVEL = 'admin'#
AUTHOR_LEVEL = 'author'#
exception DoesNotExist#

Bases: ObjectDoesNotExist

exception MultipleObjectsReturned#

Bases: MultipleObjectsReturned

READ_LEVEL = 'read'#
access_level#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_access_level_display(*, field=<django.db.models.fields.CharField: access_level>)#
group#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

group_id#
id#

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

library#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

library_id#
objects = <django.db.models.manager.Manager object>#
save(*args, **kwargs)#

Validate any constraints on the model.

We can remove this and replace it with a proper database constraint once we’re upgraded to Django 2.2+

user#

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

user_id#

openedx.core.djangoapps.content_libraries.permissions module#

Permissions for Content Libraries (v2, openedx_content-based)

Deprecated: The legacy permission rules and constants that rely on ContentLibraryPermission are deprecated in favor of openedx-authz. See openedx/openedx-platform#37409.

class openedx.core.djangoapps.content_libraries.permissions.HasPermissionInContentLibraryScope(permission: PermissionData, filter_keys: list[str] | None = None)#

Bases: Rule

Bridgekeeper rule that checks content library permissions via the openedx-authz system.

This rule integrates the openedx-authz authorization system (backed by Casbin) with Bridgekeeper’s declarative permission system. It checks if a user has been granted a specific permission (action) through their role assignments in the authorization system.

The rule works by: 1. Querying the authorization system to find library scopes where the user has this permission 2. Parsing the library keys (org/slug) from the scopes 3. Building database filters to match ContentLibrary models with those org/slug combinations

permission#

The permission object representing the action to check (e.g., ‘view’, ‘edit’). This is used to look up scopes in the authorization system.

Type:

PermissionData

filter_keys#

The Django model fields to use when building QuerySet filters. Defaults to [‘org’, ‘slug’] for ContentLibrary models.

These fields are used to construct the Q object filters that match libraries based on the parsed components from library keys in authorization scopes.

For ContentLibrary, library keys have the format ‘lib:ORG:SLUG’, which maps to: - ‘org’ -> filters on org__short_name (related Organization model) - ‘slug’ -> filters on slug field

If filtering by different fields is needed, pass a custom list. For example: - [‘org’, ‘slug’] - default for ContentLibrary (filters by org and slug) - [‘id’] - filter by primary key (for other models)

Type:

list[str]

Examples

Basic usage with default filter_keys:
>>> from bridgekeeper import perms
>>> from openedx.core.djangoapps.content_libraries.permissions import HasPermissionInContentLibraryScope
>>>
>>> # Uses default filter_keys=['org', 'slug'] for ContentLibrary
>>> can_view = HasPermissionInContentLibraryScope('view_library')
>>> perms['libraries.view_library'] = can_view
Compound permissions with boolean operators:
>>> from bridgekeeper.rules import Attribute
>>>
>>> is_active = Attribute('is_active', True)
>>> is_staff = Attribute('is_staff', True)
>>> can_view = HasPermissionInContentLibraryScope('view_library')
>>>
>>> # User must be active AND (staff OR have explicit permission)
>>> perms['libraries.view_library'] = is_active & (is_staff | can_view)
QuerySet filtering (efficient, database-level):
>>> from openedx.core.djangoapps.content_libraries.models import ContentLibrary
>>>
>>> # Gets all libraries user can view in a single SQL query
>>> visible_libraries = perms['libraries.view_library'].filter(
...     request.user,
...     ContentLibrary.objects.all()
... )
Individual object checks:
>>> library = ContentLibrary.objects.get(org__short_name='DemoX', slug='CSPROB')
>>> if perms['libraries.view_library'].check(request.user, library):
...     # User can view this specific library

Note

The library keys in authorization scopes must have the format ‘lib:ORG:SLUG’ to match the ContentLibrary model’s org.short_name and slug fields. For example, scope ‘lib:DemoX:CSPROB’ matches a library with org.short_name=’DemoX’ and slug=’CSPROB’.

check(user, instance, *args, **kwargs)#

Check if user has permission for a specific object instance.

This method is used for checking permission on individual objects rather than filtering a QuerySet. It extracts the scope from the object and checks if the user has the required permission in that scope via Casbin.

Parameters:

  • user – The Django user object (must have a ‘username’ attribute).

  • instance – The Django model instance to check permission for.

  • *args – Additional positional arguments (for compatibility with parent signature).

  • **kwargs – Additional keyword arguments (for compatibility with parent signature).

Returns:

True if the user has the permission in the object’s scope,

False otherwise.

Return type:

bool

Example

>>> rule = HasPermissionInContentLibraryScope('view')
>>> can_view = rule.check(user, library)
>>> # Checks if user has 'view' permission in scope 'lib:DemoX:CSPROB'
query(user)#

Convert this rule to a Django Q object for QuerySet filtering.

Parameters:

user – The Django user object (must have a ‘username’ attribute).

Returns:

A Django Q object that can be used to filter a QuerySet.

The Q object combines multiple conditions using OR (|) operators, where each condition matches a library’s org and slug fields: Q(org__short_name=’OrgA’ & slug=’lib-a’) | Q(org__short_name=’OrgB’ & slug=’lib-b’)

Return type:

Q

Example

>>> # User has 'view' permission in scopes: ['lib:OrgA:lib-a', 'lib:OrgB:lib-b']
>>> rule = HasPermissionInContentLibraryScope('view', filter_keys=['org', 'slug'])
>>> q = rule.query(user)
>>> # Results in: Q(org__short_name='OrgA', slug='lib-a') | Q(org__short_name='OrgB', slug='lib-b')
>>>
>>> # Apply to queryset
>>> libraries = ContentLibrary.objects.filter(q)
>>> # SQL: SELECT * FROM content_library
>>> #      WHERE (org.short_name='OrgA' AND slug='lib-a')
>>> #         OR (org.short_name='OrgB' AND slug='lib-b')

openedx.core.djangoapps.content_libraries.signal_handlers module#

Content library signal handlers.

openedx.core.djangoapps.content_libraries.signal_handlers.collection_updated(learning_package: LearningPackageEventData, change: CollectionChangeData, **kwargs) → None#

A Collection has been updated - handle that as needed.

We receive this low-level event from openedx_content, and check if it happened in a library. If so, we emit more detailed library-specific events.

⏳ This event is emitted synchronously and this handler is called

synchronously. If multiple entities were changed, we need to dispatch an asynchronous handler to deal with them to avoid slowdowns.

openedx.core.djangoapps.content_libraries.signal_handlers.entities_published(learning_package: LearningPackageEventData, change_log: PublishLogEventData, **kwargs) → None#

Entities (containers/components) have been published - handle that as needed.

We receive this low-level event from openedx_content, and check if it happened in a library. If so, we emit more detailed library-specific events.

This event change log includes entities that were directly published as well as other things that are affected as publish “side effects”.

💾 This event is only received after the transaction has committed. ⏳ This event is emitted synchronously and this handler is called

synchronously. If multiple entities were published, we need to dispatch an asynchronous handler to deal with them to avoid slowdowns. If only one entity was published, we want to deal with that synchronously so that we can show the user correct data when the current requests completes.

openedx.core.djangoapps.content_libraries.signal_handlers.entities_updated(learning_package: LearningPackageEventData, change_log: DraftChangeLogEventData, **kwargs) → None#

Entities (containers/components) have been changed - handle that as needed.

We receive this low-level event from openedx_content, and check if it happened in a library. If so, we emit more detailed library-specific events.

This event change log includes entities that were directly edited as well as their dependencies which may be only indirectly affected.

💾 This event is only received after the transaction has committed. ⏳ This event is emitted synchronously and this handler is called

synchronously. If multiple entities were changed, we need to dispatch an asynchronous handler to deal with them to avoid slowdowns. If only one entity is changed, we want to deal with that synchronously so that we can show the user correct data when the current requests completes.

openedx.core.djangoapps.content_libraries.tasks module#

Celery tasks for Content Libraries.

Architecture note:

Several functions in this file manage the copying/updating of blocks in modulestore and openedx_content. These operations should only be performed within the context of CMS. However, due to existing edx-platform code structure, we’ve had to define the functions in shared source tree (openedx/) and the tasks are registered in both LMS and CMS.

To ensure that we’re not accidentally importing things from openedx_content in the LMS context, we use ensure_cms throughout this module.

A longer-term solution to this issue would be to move the content_libraries app to cms: openedx/edx-platform#33428

class openedx.core.djangoapps.content_libraries.tasks.LibraryBackupTask#

Bases: UserTask

Base class for tasks related with Library backup functionality.

NAME_PREFIX = 'Library Learning Package Backup'#
classmethod generate_name(arguments_dict) → str#

Create a name for this particular backup task instance.

Should be both: a. semi human-friendly b. something we can query in order to determine whether the library has a task in progress

Parameters:

arguments_dict (dict) – The arguments given to the task function

Returns:

The generated name

Return type:

str

ignore_result = False#

If enabled the worker won’t store task state and return values for this task. Defaults to the :setting:`task_ignore_result` setting.

priority = None#

Default task priority.

rate_limit = None#

None (no rate limit), ‘100/s’ (hundred tasks a second), ‘100/m’ (hundred tasks a minute),`’100/h’` (hundred tasks an hour)

Type:

Rate limit for this task type. Examples

reject_on_worker_lost = None#

Even if acks_late is enabled, the worker will acknowledge tasks when the worker process executing them abruptly exits or is signaled (e.g., :sig:`KILL`/:sig:`INT`, etc).

Setting this to true allows the message to be re-queued instead, so that the task will execute again by the same worker, or another worker.

Warning: Enabling this can cause message loops; make sure you know what you’re doing.

request_stack = <celery.utils.threads._LocalStack object>#

Task request stack, the current request will be the topmost.

serializer = 'json'#

The name of a serializer that are registered with kombu.serialization.registry. Default is ‘json’.

store_errors_even_if_ignored = True#

When enabled errors will be stored even if the task is otherwise configured to ignore results.

track_started = True#

If enabled the task will report its status as ‘started’ when the task is executed by a worker. Disabled by default as the normal behavior is to not report that level of granularity. Tasks are either pending, finished, or waiting to be retried.

Having a ‘started’ status can be useful for when there are long running tasks and there’s a need to report what task is currently running.

The application default can be overridden using the :setting:`task_track_started` setting.

typing = True#

Enable argument checking. You can set this to false if you don’t want the signature to be checked when calling the task. Defaults to app.strict_typing.

exception openedx.core.djangoapps.content_libraries.tasks.LibraryRestoreLoadError(message, logfile=None)#

Bases: Exception

class openedx.core.djangoapps.content_libraries.tasks.LibraryRestoreTask#

Bases: UserTask

Base class for library restore tasks.

ARTIFACT_NAMES = {'Failed': 'Error log', 'Succeeded': 'Library Restore'}#
ERROR_LOG_ARTIFACT_NAME = 'Error log'#
NAME_PREFIX = 'Library Learning Package Restore'#
fail_with_error_log(logfile) → None#

Helper method to create an error log artifact and fail the task.

Parameters:

logfile (io.StringIO) – The error log content

classmethod generate_name(arguments_dict)#

Generate a name for the corresponding UserTaskStatus model instance.

Should be implemented by each subclass to generate a meaningful name from the task parameters. Defaults to the name of the task function.

ignore_result = False#

If enabled the worker won’t store task state and return values for this task. Defaults to the :setting:`task_ignore_result` setting.

load_learning_package(storage_path, user)#

Load learning package from a backup file in storage.

Parameters:

storage_path (str) – The path to the backup file in storage

Returns:

The result of loading the learning package, including status and info

Return type:

dict

Raises:

LibraryRestoreLoadError – If there is an error loading the learning package

priority = None#

Default task priority.

rate_limit = None#

None (no rate limit), ‘100/s’ (hundred tasks a second), ‘100/m’ (hundred tasks a minute),`’100/h’` (hundred tasks an hour)

Type:

Rate limit for this task type. Examples

reject_on_worker_lost = None#

Even if acks_late is enabled, the worker will acknowledge tasks when the worker process executing them abruptly exits or is signaled (e.g., :sig:`KILL`/:sig:`INT`, etc).

Setting this to true allows the message to be re-queued instead, so that the task will execute again by the same worker, or another worker.

Warning: Enabling this can cause message loops; make sure you know what you’re doing.

request_stack = <celery.utils.threads._LocalStack object>#

Task request stack, the current request will be the topmost.

serializer = 'json'#

The name of a serializer that are registered with kombu.serialization.registry. Default is ‘json’.

store_errors_even_if_ignored = True#

When enabled errors will be stored even if the task is otherwise configured to ignore results.

track_started = True#

If enabled the task will report its status as ‘started’ when the task is executed by a worker. Disabled by default as the normal behavior is to not report that level of granularity. Tasks are either pending, finished, or waiting to be retried.

Having a ‘started’ status can be useful for when there are long running tasks and there’s a need to report what task is currently running.

The application default can be overridden using the :setting:`task_track_started` setting.

typing = True#

Enable argument checking. You can set this to false if you don’t want the signature to be checked when calling the task. Defaults to app.strict_typing.

class openedx.core.djangoapps.content_libraries.tasks.LibrarySyncChildrenTask#

Bases: UserTask

Base class for tasks which operate upon library_content children.

classmethod generate_name(arguments_dict) → str#

Create a name for this particular import task instance.

Should be both: a. semi human-friendly b. something we can query in order to determine whether the dest block has a task in progress

Parameters:

arguments_dict (dict) – The arguments given to the task function

ignore_result = False#

If enabled the worker won’t store task state and return values for this task. Defaults to the :setting:`task_ignore_result` setting.

priority = None#

Default task priority.

rate_limit = None#

None (no rate limit), ‘100/s’ (hundred tasks a second), ‘100/m’ (hundred tasks a minute),`’100/h’` (hundred tasks an hour)

Type:

Rate limit for this task type. Examples

reject_on_worker_lost = None#

Even if acks_late is enabled, the worker will acknowledge tasks when the worker process executing them abruptly exits or is signaled (e.g., :sig:`KILL`/:sig:`INT`, etc).

Setting this to true allows the message to be re-queued instead, so that the task will execute again by the same worker, or another worker.

Warning: Enabling this can cause message loops; make sure you know what you’re doing.

request_stack = <celery.utils.threads._LocalStack object>#

Task request stack, the current request will be the topmost.

serializer = 'json'#

The name of a serializer that are registered with kombu.serialization.registry. Default is ‘json’.

store_errors_even_if_ignored = True#

When enabled errors will be stored even if the task is otherwise configured to ignore results.

track_started = True#

If enabled the task will report its status as ‘started’ when the task is executed by a worker. Disabled by default as the normal behavior is to not report that level of granularity. Tasks are either pending, finished, or waiting to be retried.

Having a ‘started’ status can be useful for when there are long running tasks and there’s a need to report what task is currently running.

The application default can be overridden using the :setting:`task_track_started` setting.

typing = True#

Enable argument checking. You can set this to false if you don’t want the signature to be checked when calling the task. Defaults to app.strict_typing.

openedx.core.djangoapps.content_libraries.tasks.dispatch_and_wait(task_fn: Task, wait_for_full_completion: bool = False, **kwargs) → None#

Try to wait for the given celery task to complete before returning, up to some reasonable timeout, and then finish anything remaining work asynchonrously.

Note: we’re not using async python, so this function will unfortunately block the current CMS worker for a few seconds.

Usage example#

Instead of:

tasks.send_change_events_for_modified_entities.delay(...)

Do:

dispatch_and_wait(
    tasks.send_change_events_for_modified_entities,
    ...
)

The wait_for_full_completion param is to simplify a common pattern. When it’s True, this will just call the function directly (not using celery) and wait indefinitely for it to complete. When it’s False, we’ll dispatch the task using celery and wait up to a given timeout. So you should set it True if you are fairly certain the task will be able to complete quickly (e.g. when processing a small number of changes).

openedx.core.djangoapps.content_libraries.tasks.emit_collections_updated(library: ContentLibrary, entity_ids: list[ID]) → None#

Helper function to notify affected collections after an entity is deleted/un-deleted/published/un-published.

Used by send_change_events_for_modified_entities() and send_events_after_publish()

openedx.core.djangoapps.content_libraries.urls module#

Module contents#

Contents