diff --git a/Makefile b/Makefile index 646870e09..7997a9270 100644 --- a/Makefile +++ b/Makefile @@ -21,6 +21,7 @@ help: ## display this help message quality: ## check coding style with pycodestyle and pylint pycodestyle + mypy pylint xblock validate: test diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 000000000..9435d230b --- /dev/null +++ b/mypy.ini @@ -0,0 +1,10 @@ +[mypy] +follow_imports = normal +ignore_missing_imports = False +allow_untyped_globals = False +files = + xblock + +# Ignore web_fragments typing until it has hints. +[mypy-web_fragments.*] +ignore_missing_imports = True diff --git a/requirements/base.txt b/requirements/base.txt index 7d289a039..7e306bdb4 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -12,7 +12,7 @@ fs==2.4.16 # via -r requirements/base.in lxml==5.1.0 # via -r requirements/base.in -mako==1.3.0 +mako==1.3.2 # via -r requirements/base.in markupsafe==2.1.4 # via @@ -24,7 +24,7 @@ pymongo==3.13.0 # via edx-opaque-keys python-dateutil==2.8.2 # via -r requirements/base.in -pytz==2023.3.post1 +pytz==2023.4 # via -r requirements/base.in pyyaml==6.0.1 # via -r requirements/base.in diff --git a/requirements/ci.txt b/requirements/ci.txt index 3dc740abb..5c0a64f65 100644 --- a/requirements/ci.txt +++ b/requirements/ci.txt @@ -22,7 +22,7 @@ packaging==23.2 # via # pyproject-api # tox -platformdirs==4.1.0 +platformdirs==4.2.0 # via # tox # virtualenv diff --git a/requirements/dev.txt b/requirements/dev.txt index 1038da687..b590dd1b5 100644 --- a/requirements/dev.txt +++ b/requirements/dev.txt @@ -21,11 +21,11 @@ attrs==23.2.0 # via # -r requirements/test.txt # hypothesis -boto3==1.34.29 +boto3==1.34.31 # via # -r requirements/test.txt # fs-s3fs -botocore==1.34.29 +botocore==1.34.31 # via # -r requirements/test.txt # boto3 @@ -56,7 +56,7 @@ click-log==0.4.0 # via # -r requirements/test.txt # edx-lint -code-annotations==1.5.0 +code-annotations==1.6.0 # via # -r requirements/test.txt # edx-lint @@ -90,7 +90,15 @@ django==2.2.28 # via # -c https://raw.githubusercontent.com/edx/edx-lint/master/edx_lint/files/common_constraints.txt # -r requirements/test.txt + # django-stubs + # django-stubs-ext # openedx-django-pyfs +django-stubs==4.2.7 + # via -r requirements/test.txt +django-stubs-ext==4.2.7 + # via + # -r requirements/test.txt + # django-stubs edx-lint==5.3.6 # via -r requirements/test.txt edx-opaque-keys==2.5.1 @@ -115,7 +123,7 @@ fs-s3fs==1.1.1 # via # -r requirements/test.txt # openedx-django-pyfs -hypothesis==6.97.1 +hypothesis==6.97.3 # via -r requirements/test.txt importlib-metadata==7.0.1 # via @@ -152,7 +160,9 @@ lazy==1.6 # via -r requirements/test.txt lxml==5.1.0 # via -r requirements/test.txt -mako==1.3.0 +lxml-stubs==0.5.1 + # via -r requirements/test.txt +mako==1.3.2 # via -r requirements/test.txt markupsafe==2.1.4 # via @@ -165,6 +175,12 @@ mccabe==0.7.0 # pylint mock==5.1.0 # via -r requirements/test.txt +mypy==1.8.0 + # via -r requirements/test.txt +mypy-extensions==1.0.0 + # via + # -r requirements/test.txt + # mypy openedx-django-pyfs==3.4.1 # via -r requirements/test.txt packaging==23.2 @@ -184,7 +200,7 @@ pbr==6.0.0 # stevedore pip-tools==7.3.0 # via -r requirements/pip-tools.txt -platformdirs==4.1.0 +platformdirs==4.2.0 # via # -r requirements/ci.txt # -r requirements/test.txt @@ -200,11 +216,11 @@ pluggy==1.4.0 # tox pycodestyle==2.11.1 # via -r requirements/test.txt -pydantic==2.5.3 +pydantic==2.6.0 # via # -r requirements/test.txt # inflect -pydantic-core==2.14.6 +pydantic-core==2.16.1 # via # -r requirements/test.txt # pydantic @@ -252,17 +268,17 @@ pytest==8.0.0 # pytest-django pytest-cov==4.1.0 # via -r requirements/test.txt -pytest-django==4.7.0 +pytest-django==4.8.0 # via -r requirements/test.txt python-dateutil==2.8.2 # via # -r requirements/test.txt # botocore -python-slugify==8.0.2 +python-slugify==8.0.3 # via # -r requirements/test.txt # code-annotations -pytz==2023.3.post1 +pytz==2023.4 # via # -r requirements/test.txt # django @@ -307,6 +323,8 @@ tomli==2.0.1 # -r requirements/test.txt # build # coverage + # django-stubs + # mypy # pip-tools # pylint # pyproject-api @@ -321,13 +339,30 @@ tox==4.12.1 # via # -r requirements/ci.txt # -r requirements/test.txt +types-python-dateutil==2.8.19.20240106 + # via -r requirements/test.txt +types-pytz==2023.4.0.20240130 + # via + # -r requirements/test.txt + # django-stubs +types-pyyaml==6.0.12.12 + # via + # -r requirements/test.txt + # django-stubs +types-setuptools==69.0.0.20240125 + # via -r requirements/test.txt +types-webob==1.8.0.20240128 + # via -r requirements/test.txt typing-extensions==4.9.0 # via # -r requirements/test.txt # annotated-types # astroid + # django-stubs + # django-stubs-ext # edx-opaque-keys # inflect + # mypy # pydantic # pydantic-core # pylint diff --git a/requirements/django.txt b/requirements/django.txt index 93e73cfa7..4d2d95584 100644 --- a/requirements/django.txt +++ b/requirements/django.txt @@ -8,9 +8,9 @@ appdirs==1.4.4 # via # -r requirements/base.txt # fs -boto3==1.34.29 +boto3==1.34.31 # via fs-s3fs -botocore==1.34.29 +botocore==1.34.31 # via # boto3 # s3transfer @@ -36,7 +36,7 @@ lazy==1.6 # via -r requirements/django.in lxml==5.1.0 # via -r requirements/base.txt -mako==1.3.0 +mako==1.3.2 # via -r requirements/base.txt markupsafe==2.1.4 # via @@ -56,7 +56,7 @@ python-dateutil==2.8.2 # via # -r requirements/base.txt # botocore -pytz==2023.3.post1 +pytz==2023.4 # via # -r requirements/base.txt # django diff --git a/requirements/doc.txt b/requirements/doc.txt index 4c606561c..f6358f508 100644 --- a/requirements/doc.txt +++ b/requirements/doc.txt @@ -18,11 +18,11 @@ babel==2.14.0 # sphinx beautifulsoup4==4.12.3 # via pydata-sphinx-theme -boto3==1.34.29 +boto3==1.34.31 # via # -r requirements/django.txt # fs-s3fs -botocore==1.34.29 +botocore==1.34.31 # via # -r requirements/django.txt # boto3 @@ -68,7 +68,7 @@ lazy==1.6 # via -r requirements/django.txt lxml==5.1.0 # via -r requirements/django.txt -mako==1.3.0 +mako==1.3.2 # via -r requirements/django.txt markupsafe==2.1.4 # via @@ -102,7 +102,7 @@ python-dateutil==2.8.2 # via # -r requirements/django.txt # botocore -pytz==2023.3.post1 +pytz==2023.4 # via # -r requirements/django.txt # babel diff --git a/requirements/test.in b/requirements/test.in index 4a5d2a57b..39e924dd3 100644 --- a/requirements/test.in +++ b/requirements/test.in @@ -8,9 +8,12 @@ astroid coverage ddt diff-cover >= 0.2.1 +django-stubs edx_lint hypothesis +lxml-stubs mock +mypy path pycodestyle pylint @@ -18,3 +21,8 @@ pytest pytest-cov pytest-django tox +types-python-dateutil +types-pytz +types-PyYAML +types-setuptools +types-WebOb diff --git a/requirements/test.txt b/requirements/test.txt index 77757534e..1e26d2bb2 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -17,11 +17,11 @@ astroid==3.0.2 # pylint-celery attrs==23.2.0 # via hypothesis -boto3==1.34.29 +boto3==1.34.31 # via # -r requirements/django.txt # fs-s3fs -botocore==1.34.29 +botocore==1.34.31 # via # -r requirements/django.txt # boto3 @@ -37,7 +37,7 @@ click==8.1.7 # edx-lint click-log==0.4.0 # via edx-lint -code-annotations==1.5.0 +code-annotations==1.6.0 # via edx-lint colorama==0.4.6 # via tox @@ -58,7 +58,13 @@ distlib==0.3.8 # via # -c https://raw.githubusercontent.com/edx/edx-lint/master/edx_lint/files/common_constraints.txt # -r requirements/django.txt + # django-stubs + # django-stubs-ext # openedx-django-pyfs +django-stubs==4.2.7 + # via -r requirements/test.in +django-stubs-ext==4.2.7 + # via django-stubs edx-lint==5.3.6 # via -r requirements/test.in edx-opaque-keys==2.5.1 @@ -80,7 +86,7 @@ fs-s3fs==1.1.1 # via # -r requirements/django.txt # openedx-django-pyfs -hypothesis==6.97.1 +hypothesis==6.97.3 # via -r requirements/test.in inflect==7.0.0 # via jinja2-pluralize @@ -104,7 +110,9 @@ lazy==1.6 # via -r requirements/django.txt lxml==5.1.0 # via -r requirements/django.txt -mako==1.3.0 +lxml-stubs==0.5.1 + # via -r requirements/test.in +mako==1.3.2 # via -r requirements/django.txt markupsafe==2.1.4 # via @@ -115,6 +123,10 @@ mccabe==0.7.0 # via pylint mock==5.1.0 # via -r requirements/test.in +mypy==1.8.0 + # via -r requirements/test.in +mypy-extensions==1.0.0 + # via mypy openedx-django-pyfs==3.4.1 # via -r requirements/django.txt packaging==23.2 @@ -128,7 +140,7 @@ pbr==6.0.0 # via # -r requirements/django.txt # stevedore -platformdirs==4.1.0 +platformdirs==4.2.0 # via # pylint # tox @@ -140,9 +152,9 @@ pluggy==1.4.0 # tox pycodestyle==2.11.1 # via -r requirements/test.in -pydantic==2.5.3 +pydantic==2.6.0 # via inflect -pydantic-core==2.14.6 +pydantic-core==2.16.1 # via pydantic pygments==2.17.2 # via diff-cover @@ -174,15 +186,15 @@ pytest==8.0.0 # pytest-django pytest-cov==4.1.0 # via -r requirements/test.in -pytest-django==4.7.0 +pytest-django==4.8.0 # via -r requirements/test.in python-dateutil==2.8.2 # via # -r requirements/django.txt # botocore -python-slugify==8.0.2 +python-slugify==8.0.3 # via code-annotations -pytz==2023.3.post1 +pytz==2023.4 # via # -r requirements/django.txt # django @@ -219,6 +231,8 @@ text-unidecode==1.3 tomli==2.0.1 # via # coverage + # django-stubs + # mypy # pylint # pyproject-api # pytest @@ -227,13 +241,30 @@ tomlkit==0.12.3 # via pylint tox==4.12.1 # via -r requirements/test.in +types-python-dateutil==2.8.19.20240106 + # via -r requirements/test.in +types-pytz==2023.4.0.20240130 + # via + # -r requirements/test.in + # django-stubs +types-pyyaml==6.0.12.12 + # via + # -r requirements/test.in + # django-stubs +types-setuptools==69.0.0.20240125 + # via -r requirements/test.in +types-webob==1.8.0.20240128 + # via -r requirements/test.in typing-extensions==4.9.0 # via # -r requirements/django.txt # annotated-types # astroid + # django-stubs + # django-stubs-ext # edx-opaque-keys # inflect + # mypy # pydantic # pydantic-core # pylint diff --git a/xblock/__init__.py b/xblock/__init__.py index d10628e18..e6ba45da9 100644 --- a/xblock/__init__.py +++ b/xblock/__init__.py @@ -25,6 +25,6 @@ def __init__(self, *args, **kwargs): # For backwards compatibility, provide the XBlockMixin in xblock.fields # without causing a circular import -xblock.fields.XBlockMixin = XBlockMixin +xblock.fields.XBlockMixin = XBlockMixin # type: ignore[attr-defined] __version__ = '1.10.0' diff --git a/xblock/core.py b/xblock/core.py index 2c93195ac..3fc3608cb 100644 --- a/xblock/core.py +++ b/xblock/core.py @@ -3,18 +3,24 @@ This code is in the Runtime layer, because it is authored once by edX and used by all runtimes. - """ -from collections import defaultdict +from __future__ import annotations + import inspect import os import warnings +import typing as t +from collections import defaultdict import pkg_resources +from lxml import etree +from opaque_keys.edx.keys import LearningContextKey, UsageKey +from web_fragments.fragment import Fragment import xblock.exceptions from xblock.exceptions import DisallowedFileError -from xblock.fields import String, List, Scope +from xblock.fields import String, List, Scope, ScopeIds +from xblock.field_data import FieldData from xblock.internal import class_lazy import xblock.mixins from xblock.mixins import ( @@ -29,6 +35,10 @@ from xblock.plugin import Plugin from xblock.validation import Validation + +if t.TYPE_CHECKING: + from xblock.runtime import Runtime + # exposing XML_NAMESPACES as a member of core, in order to avoid importing mixins where # XML_NAMESPACES are needed (e.g. runtime.py). XML_NAMESPACES = xblock.mixins.XML_NAMESPACES @@ -53,26 +63,26 @@ class SharedBlockBase(Plugin): Behaviors and attrs which all XBlock like things should share """ - resources_dir = '' - public_dir = 'public' - i18n_js_namespace = None + resources_dir: str = '' + public_dir: str = 'public' + i18n_js_namespace: str | None = None @classmethod - def get_resources_dir(cls): + def get_resources_dir(cls) -> str: """ Gets the resource directory for this XBlock. """ return cls.resources_dir @classmethod - def get_public_dir(cls): + def get_public_dir(cls) -> str: """ Gets the public directory for this XBlock. """ return cls.public_dir @classmethod - def get_i18n_js_namespace(cls): + def get_i18n_js_namespace(cls) -> str | None: """ Gets the JavaScript translations namespace for this XBlock. @@ -83,7 +93,7 @@ def get_i18n_js_namespace(cls): return cls.i18n_js_namespace @classmethod - def open_local_resource(cls, uri): + def open_local_resource(cls, uri: str) -> t.IO[bytes]: """ Open a local resource. @@ -125,25 +135,27 @@ def open_local_resource(cls, uri): # -- Base Block class XBlock(XmlSerializationMixin, HierarchyMixin, ScopedStorageMixin, RuntimeServicesMixin, HandlersMixin, IndexInfoMixin, ViewsMixin, SharedBlockBase): - """Base class for XBlocks. + """ + Base class for XBlocks. Derive from this class to create a new kind of XBlock. There are no required methods, but you will probably need at least one view. Don't provide the ``__init__`` method when deriving from this class. - """ - entry_point = 'xblock.v1' + entry_point: str = 'xblock.v1' name = String(help="Short name for the block", scope=Scope.settings) - tags = List(help="Tags for this block", scope=Scope.settings) + tags: List[str] = List(help="Tags for this block", scope=Scope.settings) + + has_children: bool @class_lazy - def _class_tags(cls): # pylint: disable=no-self-argument + def _class_tags(cls: type[XBlock]) -> set[str]: # type: ignore[misc] """ Collect the tags from all base classes. """ - class_tags = set() + class_tags: set[str] = set() for base in cls.mro()[1:]: # pylint: disable=no-member class_tags.update(getattr(base, '_class_tags', set())) @@ -153,7 +165,7 @@ def _class_tags(cls): # pylint: disable=no-self-argument @staticmethod def tag(tags): """Returns a function that adds the words in `tags` as class tags to this class.""" - def dec(cls): + def dec(cls: type[XBlock]) -> type[XBlock]: """Add the words in `tags` as class tags to this class.""" # Add in this class's tags cls._class_tags.update(tags.replace(",", " ").split()) # pylint: disable=protected-access @@ -161,7 +173,9 @@ def dec(cls): return dec @classmethod - def load_tagged_classes(cls, tag, fail_silently=True): + def load_tagged_classes( + cls, tag: str, fail_silently: bool = True + ) -> t.Iterable[tuple[str, type[XBlock]]]: """ Produce a sequence of all XBlock classes tagged with `tag`. @@ -174,14 +188,20 @@ def load_tagged_classes(cls, tag, fail_silently=True): (e.g. on startup or first page load), and in what contexts. Hence, the flag. """ - # Allow this method to access the `_class_tags` - # pylint: disable=W0212 for name, class_ in cls.load_classes(fail_silently): - if tag in class_._class_tags: - yield name, class_ + xblock_class: type[XBlock] = class_ # type: ignore + if tag in xblock_class._class_tags: # pylint: disable=protected-access + yield name, xblock_class # pylint: disable=keyword-arg-before-vararg - def __init__(self, runtime, field_data=None, scope_ids=UNSET, *args, **kwargs): + def __init__( + self, + runtime: Runtime, + field_data: FieldData | None = None, + scope_ids: ScopeIds | object = UNSET, + *args, + **kwargs + ): """ Construct a new XBlock. @@ -201,12 +221,14 @@ def __init__(self, runtime, field_data=None, scope_ids=UNSET, *args, **kwargs): """ if scope_ids is UNSET: raise TypeError('scope_ids are required') + else: + assert isinstance(scope_ids, ScopeIds) # Provide backwards compatibility for external access through _field_data super().__init__(runtime=runtime, scope_ids=scope_ids, field_data=field_data, *args, **kwargs) @property - def usage_key(self): + def usage_key(self) -> UsageKey: """ A key identifying this particular usage of the XBlock, unique across all learning contexts in the system. @@ -215,7 +237,7 @@ def usage_key(self): return self.scope_ids.usage_id @property - def context_key(self): + def context_key(self) -> LearningContextKey | None: """ A key identifying the learning context (course, library, etc.) that contains this XBlock usage. @@ -231,11 +253,11 @@ def context_key(self): """ return getattr(self.scope_ids.usage_id, "context_key", None) - def render(self, view, context=None): + def render(self, view: str, context: dict | None = None) -> Fragment: """Render `view` with this block's runtime and the supplied `context`""" return self.runtime.render(self, view, context) - def validate(self): + def validate(self) -> Validation: """ Ask this xblock to validate itself. Subclasses are expected to override this method, as there is currently only a no-op implementation. Any overriding method @@ -244,7 +266,7 @@ def validate(self): """ return Validation(self.scope_ids.usage_id) - def ugettext(self, text): + def ugettext(self, text) -> str: """ Translates message/text and returns it in a unicode string. Using runtime to get i18n service. @@ -253,7 +275,7 @@ def ugettext(self, text): runtime_ugettext = runtime_service.ugettext return runtime_ugettext(text) - def add_xml_to_node(self, node): + def add_xml_to_node(self, node: etree._Element) -> None: """ For exporting, set data on etree.Element `node`. """ @@ -262,15 +284,19 @@ def add_xml_to_node(self, node): self.add_children_to_node(node) +# An XBlockAside's view method takes itself, an XBlock, and optional context dict, +# and returns a Fragment. +AsideView = t.Callable[["XBlockAside", XBlock, t.Optional[dict]], Fragment] + + class XBlockAside(XmlSerializationMixin, ScopedStorageMixin, RuntimeServicesMixin, HandlersMixin, SharedBlockBase): """ This mixin allows Xblock-like class to declare that it provides aside functionality. """ - - entry_point = "xblock_asides.v1" + entry_point: str = "xblock_asides.v1" @classmethod - def aside_for(cls, view_name): + def aside_for(cls, view_name: str) -> t.Callable[[AsideView], AsideView]: """ A decorator to indicate a function is the aside view for the given view_name. @@ -283,16 +309,16 @@ def student_aside(self, block, context=None): """ # pylint: disable=protected-access - def _decorator(func): + def _decorator(func: AsideView) -> AsideView: if not hasattr(func, '_aside_for'): - func._aside_for = [] + func._aside_for = [] # type: ignore - func._aside_for.append(view_name) # pylint: disable=protected-access + func._aside_for.append(view_name) # type: ignore return func return _decorator @classmethod - def should_apply_to_block(cls, block): # pylint: disable=unused-argument + def should_apply_to_block(cls, block: XBlock) -> bool: # pylint: disable=unused-argument """ Return True if the aside should be applied to a given block. This can be overridden if some aside should only wrap blocks with certain properties. @@ -300,39 +326,33 @@ def should_apply_to_block(cls, block): # pylint: disable=unused-argument return True @class_lazy - def _combined_asides(cls): # pylint: disable=no-self-argument + def _combined_asides(cls) -> dict[str, str | None]: # pylint: disable=no-self-argument """ A dictionary mapping XBlock view names to the aside method that decorates them (or None, if there is no decorator for the specified view). """ # The method declares what views it decorates. We rely on `dir` # to handle subclasses and overrides. - combined_asides = defaultdict(None) + combined_asides: dict[str, str | None] = defaultdict(None) for _view_name, view_func in inspect.getmembers(cls, lambda attr: hasattr(attr, '_aside_for')): aside_for = getattr(view_func, '_aside_for', []) for view in aside_for: combined_asides[view] = view_func.__name__ return combined_asides - def aside_view_declaration(self, view_name): + def aside_view_declaration(self, view_name: str) -> AsideView | None: """ Find and return a function object if one is an aside_view for the given view_name Aside methods declare their view provision via @XBlockAside.aside_for(view_name) This function finds those declarations for a block. - - Arguments: - view_name (str): the name of the view requested. - - Returns: - either the function or None """ if view_name in self._combined_asides: # pylint: disable=unsupported-membership-test return getattr(self, self._combined_asides[view_name]) # pylint: disable=unsubscriptable-object else: return None - def needs_serialization(self): + def needs_serialization(self) -> bool: """ Return True if the aside has any data to serialize to XML. diff --git a/xblock/django/request.py b/xblock/django/request.py index 9e17e6721..6c27f05d3 100644 --- a/xblock/django/request.py +++ b/xblock/django/request.py @@ -1,16 +1,21 @@ """Helpers for WebOb requests and responses.""" +from __future__ import annotations + from collections.abc import MutableMapping from itertools import chain, repeat from lazy import lazy +import typing as t import webob -from django.http import HttpResponse -from webob.multidict import MultiDict, NestedMultiDict, NoVars +import webob.multidict +import django.core.files.uploadedfile +import django.http +import django.utils.datastructures -def webob_to_django_response(webob_response): +def webob_to_django_response(webob_response: webob.Response) -> django.http.HttpResponse: """Returns a django response to the `webob_response`""" - django_response = HttpResponse( + django_response = django.http.HttpResponse( webob_response.app_iter, content_type=webob_response.content_type, status=webob_response.status_code, @@ -28,11 +33,11 @@ class HeaderDict(MutableMapping): """ UNPREFIXED_HEADERS = ('CONTENT_TYPE', 'CONTENT_LENGTH') - def __init__(self, meta): + def __init__(self, meta: dict[str, str]): super().__init__() self._meta = meta - def _meta_name(self, name): + def _meta_name(self, name: str) -> str: """ Translate HTTP header names to the format used by Django request objects. @@ -43,7 +48,7 @@ def _meta_name(self, name): name = 'HTTP_' + name return name - def _un_meta_name(self, name): + def _un_meta_name(self, name: str) -> str: """ Reverse of _meta_name """ @@ -51,33 +56,35 @@ def _un_meta_name(self, name): name = name[5:] return name.replace('_', '-').title() - def __getitem__(self, name): + def __getitem__(self, name: str): return self._meta[self._meta_name(name)] - def __setitem__(self, name, value): + def __setitem__(self, name: str, value: str): self._meta[self._meta_name(name)] = value - def __delitem__(self, name): + def __delitem__(self, name: str): del self._meta[self._meta_name(name)] - def __iter__(self): + def __iter__(self) -> t.Iterator[str]: for key in self._meta: if key in self.UNPREFIXED_HEADERS or key.startswith('HTTP_'): yield self._un_meta_name(key) - def __len__(self): + def __len__(self) -> int: return len(list(self)) -def querydict_to_multidict(query_dict, wrap=None): +def querydict_to_multidict( + query_dict: django.utils.datastructures.MultiValueDict, + wrap: t.Callable[[t.Any], t.Any] | None = None +) -> webob.multidict.MultiDict: """ Returns a new `webob.MultiDict` from a `django.http.QueryDict`. If `wrap` is provided, it's used to wrap the values. - """ wrap = wrap or (lambda val: val) - return MultiDict(chain.from_iterable( + return webob.multidict.MultiDict(chain.from_iterable( zip(repeat(key), (wrap(v) for v in vals)) for key, vals in query_dict.lists() )) @@ -87,32 +94,35 @@ class DjangoUploadedFile: """ Looks like a cgi.FieldStorage, but wraps a Django UploadedFile. """ - def __init__(self, uploaded): + def __init__(self, uploaded: django.core.files.uploadedfile.UploadedFile): # FieldStorage needs a file attribute. - self.file = uploaded + self.file: t.Any = uploaded @property - def name(self): + def name(self) -> str: """The name of the input element used to upload the file.""" return self.file.field_name @property - def filename(self): + def filename(self) -> str: """The name of the uploaded file.""" return self.file.name class DjangoWebobRequest(webob.Request): """ - An implementation of the webob request api, backed - by a django request + An implementation of the webob request api, backed by a django request """ - def __init__(self, request): + # Note: + # This implementation is close enough to webob.Request for it to work OK, but it does + # make mypy complain that the type signatures are different, hence the 'type: ignore' pragmas. + + def __init__(self, request: django.http.HttpRequest): self._request = request super().__init__(self.environ) @lazy - def environ(self): + def environ(self) -> dict[str, str]: # type: ignore[override] """ Add path_info to the request's META dictionary. """ @@ -123,40 +133,40 @@ def environ(self): return environ @property - def GET(self): + def GET(self) -> webob.multidict.MultiDict: # type: ignore[override] """ Returns a new `webob.MultiDict` from the request's GET query. """ return querydict_to_multidict(self._request.GET) @property - def POST(self): + def POST(self) -> webob.multidict.MultiDict | webob.multidict.NoVars: # type: ignore[override] if self.method not in ('POST', 'PUT', 'PATCH'): - return NoVars('Not a form request') + return webob.multidict.NoVars('Not a form request') # Webob puts uploaded files into the POST dictionary, so here we # combine the Django POST data and uploaded FILES data into a single # dict. - return NestedMultiDict( + return webob.multidict.NestedMultiDict( querydict_to_multidict(self._request.POST), querydict_to_multidict(self._request.FILES, wrap=DjangoUploadedFile), ) @property - def body(self): + def body(self) -> bytes: # type: ignore[override] """ Return the content of the request body. """ return self._request.body - @property - def body_file(self): + @property # type: ignore[misc] + def body_file(self) -> django.http.HttpRequest: # type: ignore[override] """ Input stream of the request """ return self._request -def django_to_webob_request(django_request): +def django_to_webob_request(django_request: django.http.HttpRequest) -> webob.Request: """Returns a WebOb request to the `django_request`""" return DjangoWebobRequest(django_request) diff --git a/xblock/fields.py b/xblock/fields.py index 8f5eecc8a..c30ac690b 100644 --- a/xblock/fields.py +++ b/xblock/fields.py @@ -5,23 +5,32 @@ for each scope. """ -from collections import namedtuple +from __future__ import annotations + import copy -import datetime import hashlib import itertools import json import re import traceback +import typing as t import warnings +from dataclasses import dataclass +from datetime import datetime, timedelta +from enum import Enum import dateutil.parser -from lxml import etree import pytz import yaml +from lxml import etree +from opaque_keys.edx.keys import DefinitionKey, UsageKey from xblock.internal import Nameable +if t.TYPE_CHECKING: + from xblock.core import XBlock + + # __all__ controls what classes end up in the docs, and in what order. __all__ = [ 'BlockScope', 'UserScope', 'Scope', 'ScopeIds', @@ -43,40 +52,36 @@ class ModifyingEnforceTypeWarning(DeprecationWarning): """ +@dataclass(frozen=True) class Sentinel: """ Class for implementing sentinel objects (only equal to themselves). """ + name: str - def __init__(self, name): - """ - `name` is the name used to identify the sentinel (which will - be displayed as the __repr__) of the sentinel. - """ - self.name = name - - def __repr__(self): + def __repr__(self) -> str: return self.name @property - def attr_name(self): + def attr_name(self) -> str: """ TODO: Look into namespace collisions. block.name_space == block_name.space """ return self.name.lower().replace('.', '_') - def __eq__(self, other): - """ Equality is based on being of the same class, and having same name + def __eq__(self, other: object) -> bool: """ - return isinstance(other, Sentinel) and self.name == other.name + Equality is based on being of the same class, and having same name + """ + return isinstance(other, self.__class__) and self.name == other.name - def __hash__(self): + def __hash__(self) -> int: """ Use a hash of the name of the sentinel """ return hash(self.name) -class BlockScope: +class BlockScope(Enum): """ Enumeration of block scopes. @@ -96,23 +101,25 @@ class BlockScope: information that is purely about the student. """ - USAGE = Sentinel('BlockScope.USAGE') - DEFINITION = Sentinel('BlockScope.DEFINITION') - TYPE = Sentinel('BlockScope.TYPE') - ALL = Sentinel('BlockScope.ALL') + USAGE = 'BlockScope.USAGE' + DEFINITION = 'BlockScope.DEFINITION' + TYPE = 'BlockScope.TYPE' + ALL = 'BlockScope.ALL' @classmethod - def scopes(cls): + def scopes(cls) -> list[BlockScope]: """ Return a list of valid/understood class scopes. """ - # Why do we need this? This should either - # * Be bubbled to the places where it is used (AcidXBlock). - # * Be automatic. Look for all members of a type. - return [cls.USAGE, cls.DEFINITION, cls.TYPE, cls.ALL] + return list(cls) + + @property + def attr_name(self) -> str: + return self.name.lower().replace('.', '_') + -class UserScope: +class UserScope(Enum): """ Enumeration of user scopes. @@ -133,22 +140,27 @@ class UserScope: submitted by all students. """ - NONE = Sentinel('UserScope.NONE') - ONE = Sentinel('UserScope.ONE') - ALL = Sentinel('UserScope.ALL') + NONE = 'UserScope.NONE' + ONE = 'UserScope.ONE' + ALL = 'UserScope.ALL' @classmethod - def scopes(cls): + def scopes(cls) -> list[UserScope]: """ Return a list of valid/understood class scopes. Why do we need this? I believe it is not used anywhere. """ - return [cls.NONE, cls.ONE, cls.ALL] + return list(cls) + @property + def attr_name(self) -> str: + return self.name.lower().replace('.', '_') -UNSET = Sentinel("fields.UNSET") -ScopeBase = namedtuple('ScopeBase', 'user block name') +class ScopeBase(t.NamedTuple): + user: UserScope + block: BlockScope + name: str class Scope(ScopeBase): @@ -184,15 +196,15 @@ class Scope(ScopeBase): the points scored by all users attempting a problem. """ - content = ScopeBase(UserScope.NONE, BlockScope.DEFINITION, 'content') - settings = ScopeBase(UserScope.NONE, BlockScope.USAGE, 'settings') - user_state = ScopeBase(UserScope.ONE, BlockScope.USAGE, 'user_state') - preferences = ScopeBase(UserScope.ONE, BlockScope.TYPE, 'preferences') - user_info = ScopeBase(UserScope.ONE, BlockScope.ALL, 'user_info') - user_state_summary = ScopeBase(UserScope.ALL, BlockScope.USAGE, 'user_state_summary') + content: t.ClassVar = ScopeBase(UserScope.NONE, BlockScope.DEFINITION, 'content') + settings: t.ClassVar = ScopeBase(UserScope.NONE, BlockScope.USAGE, 'settings') + user_state: t.ClassVar = ScopeBase(UserScope.ONE, BlockScope.USAGE, 'user_state') + preferences: t.ClassVar = ScopeBase(UserScope.ONE, BlockScope.TYPE, 'preferences') + user_info: t.ClassVar = ScopeBase(UserScope.ONE, BlockScope.ALL, 'user_info') + user_state_summary: t.ClassVar = ScopeBase(UserScope.ALL, BlockScope.USAGE, 'user_state_summary') @classmethod - def named_scopes(cls): + def named_scopes(cls) -> list[ScopeBase]: """Return all named Scopes.""" return [ cls.content, @@ -204,7 +216,7 @@ def named_scopes(cls): ] @classmethod - def scopes(cls): + def scopes(cls) -> list[ScopeBase]: """Return all possible Scopes.""" named_scopes = cls.named_scopes() return named_scopes + [ @@ -214,56 +226,98 @@ def scopes(cls): if cls(user, block) not in named_scopes ] - def __new__(cls, user, block, name=None): + def __new__(cls, user: UserScope, block, name: str | None = None) -> Scope: # type: ignore """Create a new Scope, with an optional name.""" + # TODO: This is a pretty wacky way to set a default value for `name`. + # We should try to refactor this so that Scope is just, like, + # a dataclass with a __post_init__ hook that sets the default `name`. + return ScopeBase.__new__(cls, user, block, name or f'{user}_{block}') - if name is None: - name = f'{user}_{block}' - - return ScopeBase.__new__(cls, user, block, name) - - children = Sentinel('Scope.children') - parent = Sentinel('Scope.parent') + children: t.ClassVar = Sentinel('Scope.children') + parent: t.ClassVar = Sentinel('Scope.parent') - def __str__(self): + def __str__(self) -> str: return self.name - def __eq__(self, other): + def __eq__(self, other: object) -> bool: return isinstance(other, Scope) and self.user == other.user and self.block == other.block - def __hash__(self): + def __hash__(self) -> int: return hash(('xblock.fields.Scope', self.user, self.block)) -class ScopeIds(namedtuple('ScopeIds', 'user_id block_type def_id usage_id')): +class ScopeIds(t.NamedTuple): """ A simple wrapper to collect all of the ids needed to correctly identify an XBlock (or other classes deriving from ScopedStorageMixin) to a FieldData. These identifiers match up with BlockScope and UserScope attributes, so that, for instance, the `def_id` identifies scopes that use BlockScope.DEFINITION. """ - __slots__ = () + user_id: int | str | None + block_type: str + def_id: DefinitionKey + usage_id: UsageKey + + +ScopeIds.__slots__ = () # type: ignore + + +@dataclass(frozen=True) +class Unset(Sentinel): + """ + Indicates that default value has not been provided. + """ + name: str = "fields.UNSET" + + +@dataclass(frozen=True) +class UniqueId(Sentinel): + """ + A special reference that can be used as a field's default in field + definition to signal that the field should default to a unique string value + calculated at runtime. + """ + name: str = "fields.UNIQUE_ID" -# Define special reference that can be used as a field's default in field -# definition to signal that the field should default to a unique string value -# calculated at runtime. -UNIQUE_ID = Sentinel("fields.UNIQUE_ID") +@dataclass(frozen=True) +class NoCacheValue(Sentinel): + """ + Placeholder ('nil') value to indicate when nothing has been stored + in the cache ("None" may be a valid value in the cache, so we cannot use it). + """ + name: str = "fields.NO_CACHE_VALUE" + + +@dataclass(frozen=True) +class ExplicitlySet(Sentinel): + """ + Placeholder value that indicates that a value is explicitly dirty, + because it was explicitly set. + """ + name: str = "fields.EXPLICITLY_SET" + -# define a placeholder ('nil') value to indicate when nothing has been stored -# in the cache ("None" may be a valid value in the cache, so we cannot use it). -NO_CACHE_VALUE = Sentinel("fields.NO_CACHE_VALUE") +# For backwards API compatibility, define an instance of each Field-related sentinel. +# Because our sentinels are now frozen dataclasses (i.e., value objects), these could be +# be removed in favor of just constructing the class. For example, every use of +# `UNIQUE_ID` could be replaced with `UniqueId()`. +UNSET = Unset() +UNIQUE_ID = UniqueId() +NO_CACHE_VALUE = NoCacheValue() +EXPLICITLY_SET = ExplicitlySet() -# define a placeholder value that indicates that a value is explicitly dirty, -# because it was explicitly set -EXPLICITLY_SET = Sentinel("fields.EXPLICITLY_SET") # Fields that cannot have runtime-generated defaults. These are special, # because they define the structure of XBlock trees. NO_GENERATED_DEFAULTS = ('parent', 'children') +# Type parameters of Fields. These only matter for static type analysis (mypy). +FieldValue = t.TypeVar("FieldValue") # What does the field hold? +InnerFieldValue = t.TypeVar("InnerFieldValue") # For Dict/List/Set fields: What do they contain? -class Field(Nameable): + +class Field(Nameable, t.Generic[FieldValue]): """ A field class that can be used as a class attribute to define what data the class will want to refer to. @@ -307,21 +361,30 @@ class will want to refer to. runtime_options. """ - MUTABLE = True - _default = None - # Indicates if a field's None value should be sent to the XML representation. - none_to_xml = False + MUTABLE: bool = True + _default: FieldValue | None | UniqueId = None - # We're OK redefining built-in `help` - def __init__(self, help=None, default=UNSET, scope=Scope.content, # pylint:disable=redefined-builtin - display_name=None, values=None, enforce_type=False, - xml_node=False, force_export=False, **kwargs): + # Indicates if a field's None value should be sent to the XML representation. + none_to_xml: bool = False + + def __init__( + self, + help: str | None = None, # pylint:disable=redefined-builtin + default: FieldValue | None | UniqueId | Unset = Unset(), + scope: ScopeBase = Scope.content, + display_name: str | None = None, + values: list[object] | None = None, + enforce_type: bool = False, + xml_node: bool = False, + force_export: bool = False, + **kwargs + ): self.warned = False self.help = help self._enable_enforce_type = enforce_type - if default is not UNSET: - if default is UNIQUE_ID: - self._default = UNIQUE_ID + if not isinstance(default, Unset): + if isinstance(default, UniqueId): + self._default = UniqueId() else: self._default = self._check_or_enforce_type(default) self.scope = scope @@ -332,7 +395,7 @@ def __init__(self, help=None, default=UNSET, scope=Scope.content, # pylint:disa self.force_export = force_export @property - def default(self): + def default(self) -> FieldValue | None | UniqueId: """Returns the static value that this defaults to.""" if self.MUTABLE: return copy.deepcopy(self._default) @@ -340,13 +403,13 @@ def default(self): return self._default @property - def name(self): + def name(self) -> str: """Returns the name of this field.""" # This is set by ModelMetaclass return self.__name__ or 'unknown' @property - def values(self): + def values(self) -> t.Any: """ Returns the valid values for this class. This is useful for representing possible values in a UI. @@ -379,7 +442,7 @@ def values(self): return self._values @property - def display_name(self): + def display_name(self) -> str: """ Returns the display name for this class, suitable for use in a GUI. @@ -387,27 +450,27 @@ def display_name(self): """ return self._display_name if self._display_name is not None else self.name - def _get_cached_value(self, xblock): + def _get_cached_value(self, xblock) -> FieldValue | None | NoCacheValue: """ Return a value from the xblock's cache, or a marker value if either the cache doesn't exist or the value is not found in the cache. """ - return getattr(xblock, '_field_data_cache', {}).get(self.name, NO_CACHE_VALUE) + return getattr(xblock, '_field_data_cache', {}).get(self.name, NoCacheValue()) - def _set_cached_value(self, xblock, value): + def _set_cached_value(self, xblock: XBlock, value: FieldValue | None): """Store a value in the xblock's cache, creating the cache if necessary.""" # pylint: disable=protected-access if not hasattr(xblock, '_field_data_cache'): xblock._field_data_cache = {} xblock._field_data_cache[self.name] = value - def _del_cached_value(self, xblock): + def _del_cached_value(self, xblock: XBlock): """Remove a value from the xblock's cache, if the cache exists.""" # pylint: disable=protected-access if hasattr(xblock, '_field_data_cache') and self.name in xblock._field_data_cache: del xblock._field_data_cache[self.name] - def _mark_dirty(self, xblock, value): + def _mark_dirty(self, xblock: XBlock, value: FieldValue | None | ExplicitlySet): """Set this field to dirty on the xblock.""" # pylint: disable=protected-access @@ -416,7 +479,7 @@ def _mark_dirty(self, xblock, value): if self not in xblock._dirty_fields: xblock._dirty_fields[self] = copy.deepcopy(value) - def _is_dirty(self, xblock): + def _is_dirty(self, xblock: XBlock) -> bool: """ Return whether this field should be saved when xblock.save() is called """ @@ -425,15 +488,15 @@ def _is_dirty(self, xblock): return False baseline = xblock._dirty_fields[self] - return baseline is EXPLICITLY_SET or xblock._field_data_cache[self.name] != baseline + return isinstance(baseline, ExplicitlySet) or xblock._field_data_cache[self.name] != baseline - def _is_lazy(self, value): + def _is_lazy(self, value: FieldValue | None) -> bool: """ Detect if a value is being evaluated lazily by Django. """ return 'django.utils.functional.' in str(type(value)) - def _check_or_enforce_type(self, value): + def _check_or_enforce_type(self, value: t.Any) -> FieldValue | None: """ Depending on whether enforce_type is enabled call self.enforce_type and return the result or call it and trigger a silent warning if the result @@ -462,9 +525,9 @@ def _check_or_enforce_type(self, value): value, new_value) warnings.warn(message, ModifyingEnforceTypeWarning, stacklevel=3) - return value + return value # type: ignore - def _calculate_unique_id(self, xblock): + def _calculate_unique_id(self, xblock: XBlock) -> str: """ Provide a default value for fields with `default=UNIQUE_ID`. @@ -474,7 +537,7 @@ def _calculate_unique_id(self, xblock): key = scope_key(self, xblock) return hashlib.sha1(key.encode('utf-8')).hexdigest() - def _get_default_value_to_cache(self, xblock): + def _get_default_value_to_cache(self, xblock: XBlock) -> FieldValue | None: """ Perform special logic to provide a field's default value for caching. """ @@ -482,39 +545,42 @@ def _get_default_value_to_cache(self, xblock): # pylint: disable=protected-access return self.from_json(xblock._field_data.default(xblock, self.name)) except KeyError: - if self._default is UNIQUE_ID: + if isinstance(self.default, UniqueId): return self._check_or_enforce_type(self._calculate_unique_id(xblock)) else: return self.default - def _sanitize(self, value): + def _sanitize(self, value: FieldValue | None) -> FieldValue | None: """ Allow the individual fields to sanitize the value being set -or- "get". For example, a String field wants to remove control characters. """ return value - def __get__(self, xblock, xblock_class): + def __get__(self, xblock: XBlock, xblock_class: type[XBlock]) -> FieldValue | None: """ Gets the value of this xblock. Prioritizes the cached value over obtaining the value from the field-data service. Thus if a cached value exists, that is the value that will be returned. """ if xblock is None: - return self + return self # TODO: wtf? field_data = xblock._field_data - value = self._get_cached_value(xblock) - if value is NO_CACHE_VALUE: + cache_value = self._get_cached_value(xblock) + value: FieldValue | None + if isinstance(cache_value, NoCacheValue): if field_data.has(xblock, self.name): value = self.from_json(field_data.get(xblock, self.name)) elif self.name not in NO_GENERATED_DEFAULTS: # Cache default value value = self._get_default_value_to_cache(xblock) - else: - value = self.default + else: # TODO: why aren't we handling UniqueId here??? + value = self.default # type: ignore self._set_cached_value(xblock, value) + else: + value = cache_value # If this is a mutable type, mark it as dirty, since mutations can occur without an # explicit call to __set__ (but they do require a call to __get__) @@ -523,7 +589,7 @@ def __get__(self, xblock, xblock_class): return self._sanitize(value) - def __set__(self, xblock, value): + def __set__(self, xblock: XBlock, value: FieldValue | None) -> None: """ Sets the `xblock` to the given `value`. Setting a value does not update the underlying data store; the @@ -535,7 +601,7 @@ def __set__(self, xblock, value): """ value = self._check_or_enforce_type(value) value = self._sanitize(value) - cached_value = self._get_cached_value(xblock) + cached_value = self._get_cached_value(xblock) # TODO: why aren't we handling NoCacheValue?? try: value_has_changed = cached_value != value except Exception: # pylint: disable=broad-except @@ -544,10 +610,10 @@ def __set__(self, xblock, value): value_has_changed = True if value_has_changed: # Mark the field as dirty and update the cache - self._mark_dirty(xblock, EXPLICITLY_SET) + self._mark_dirty(xblock, ExplicitlySet()) self._set_cached_value(xblock, value) - def __delete__(self, xblock): + def __delete__(self, xblock: XBlock): """ Deletes `xblock` from the underlying data store. Deletes are not cached; they are performed immediately. @@ -591,7 +657,7 @@ def _warn_deprecated_outside_JSONField(self): # pylint: disable=invalid-name ) self.warned = True - def to_json(self, value): + def to_json(self, value: FieldValue | None) -> t.Any: """ Return value in the form of nested lists and dictionaries (suitable for passing to json.dumps). @@ -602,7 +668,7 @@ def to_json(self, value): self._warn_deprecated_outside_JSONField() return value - def from_json(self, value): + def from_json(self, value) -> FieldValue | None: """ Return value as a native full featured python type (the inverse of to_json) @@ -610,22 +676,21 @@ def from_json(self, value): object """ self._warn_deprecated_outside_JSONField() - return value + return value # type: ignore - def to_string(self, value): + def to_string(self, value: FieldValue | None) -> str: """ Return a JSON serialized string representation of the value. """ self._warn_deprecated_outside_JSONField() - value = json.dumps( + return json.dumps( self.to_json(value), indent=2, sort_keys=True, separators=(',', ': '), ) - return value - def from_string(self, serialized): + def from_string(self, serialized: str) -> FieldValue | None: """ Returns a native value from a YAML serialized string representation. Since YAML is a superset of JSON, this is the inverse of to_string.) @@ -634,7 +699,7 @@ def from_string(self, serialized): value = yaml.safe_load(serialized) return self.enforce_type(value) - def enforce_type(self, value): + def enforce_type(self, value: t.Any) -> FieldValue | None: """ Coerce the type of the value, if necessary @@ -644,52 +709,52 @@ def enforce_type(self, value): This must not have side effects, since it will be executed to trigger a DeprecationWarning even if enforce_type is disabled """ - return value + return value # type: ignore - def read_from(self, xblock): + def read_from(self, xblock: XBlock) -> FieldValue | None: """ Retrieve the value for this field from the specified xblock """ return self.__get__(xblock, xblock.__class__) # pylint: disable=unnecessary-dunder-call - def read_json(self, xblock): + def read_json(self, xblock: XBlock) -> FieldValue | None: """ Retrieve the serialized value for this field from the specified xblock """ self._warn_deprecated_outside_JSONField() return self.to_json(self.read_from(xblock)) - def write_to(self, xblock, value): + def write_to(self, xblock: XBlock, value: FieldValue | None) -> None: """ Set the value for this field to value on the supplied xblock """ self.__set__(xblock, value) # pylint: disable=unnecessary-dunder-call - def delete_from(self, xblock): + def delete_from(self, xblock: XBlock) -> None: """ Delete the value for this field from the supplied xblock """ self.__delete__(xblock) # pylint: disable=unnecessary-dunder-call - def is_set_on(self, xblock): + def is_set_on(self, xblock: XBlock) -> bool: """ Return whether this field has a non-default value on the supplied xblock """ # pylint: disable=protected-access return self._is_dirty(xblock) or xblock._field_data.has(xblock, self.name) - def __hash__(self): + def __hash__(self) -> int: return hash(self.name) -class JSONField(Field): +class JSONField(Field, t.Generic[FieldValue]): """ Field type which has a convenient JSON representation. """ # for now; we'll bubble functions down when we finish deprecation in Field -class Integer(JSONField): +class Integer(JSONField[int]): """ A field that contains an integer. @@ -703,7 +768,7 @@ class Integer(JSONField): """ MUTABLE = False - def from_json(self, value): + def from_json(self, value) -> int | None: if value is None or value == '': return None return int(value) @@ -711,7 +776,7 @@ def from_json(self, value): enforce_type = from_json -class Float(JSONField): +class Float(JSONField[float]): """ A field that contains a float. @@ -722,7 +787,7 @@ class Float(JSONField): """ MUTABLE = False - def from_json(self, value): + def from_json(self, value) -> float | None: if value is None or value == '': return None return float(value) @@ -730,7 +795,7 @@ def from_json(self, value): enforce_type = from_json -class Boolean(JSONField): +class Boolean(JSONField[bool]): """ A field class for representing a boolean. @@ -759,7 +824,7 @@ def __init__(self, help=None, default=UNSET, scope=Scope.content, display_name=N values=({'display_name': "True", "value": True}, {'display_name': "False", "value": False}), **kwargs) - def from_json(self, value): + def from_json(self, value) -> bool | None: if isinstance(value, bytes): value = value.decode('ascii', errors='replace') if isinstance(value, str): @@ -770,7 +835,7 @@ def from_json(self, value): enforce_type = from_json -class Dict(JSONField): +class Dict(JSONField[t.Dict[str, InnerFieldValue]], t.Generic[InnerFieldValue]): """ A field class for representing a Python dict. @@ -779,7 +844,7 @@ class Dict(JSONField): """ _default = {} - def from_json(self, value): + def from_json(self, value) -> dict[str, InnerFieldValue] | None: if value is None or isinstance(value, dict): return value else: @@ -787,7 +852,7 @@ def from_json(self, value): enforce_type = from_json - def to_string(self, value): + def to_string(self, value) -> str: """ In python3, json.dumps() cannot sort keys of different types, so preconvert None to 'null'. @@ -800,7 +865,7 @@ def to_string(self, value): return super().to_string(value) -class List(JSONField): +class List(JSONField[t.List[InnerFieldValue]], t.Generic[InnerFieldValue]): """ A field class for representing a list. @@ -809,7 +874,7 @@ class List(JSONField): """ _default = [] - def from_json(self, value): + def from_json(self, value) -> list[InnerFieldValue] | None: if value is None or isinstance(value, list): return value else: @@ -818,7 +883,7 @@ def from_json(self, value): enforce_type = from_json -class Set(JSONField): +class Set(JSONField[t.List[InnerFieldValue]], t.Generic[InnerFieldValue]): """ A field class for representing a set. @@ -837,7 +902,7 @@ def __init__(self, *args, **kwargs): self._default = set(self._default) - def from_json(self, value): + def from_json(self, value) -> set[InnerFieldValue] | None: if value is None or isinstance(value, set): return value else: @@ -846,7 +911,7 @@ def from_json(self, value): enforce_type = from_json -class String(JSONField): +class String(JSONField[str]): """ A field class for representing a string. @@ -856,7 +921,7 @@ class String(JSONField): MUTABLE = False BAD_REGEX = re.compile('[\x00-\x08\x0b\x0c\x0e-\x1f\ud800-\udfff\ufffe\uffff]', flags=re.UNICODE) - def _sanitize(self, value): + def _sanitize(self, value) -> str | None: """ Remove the control characters that are not allowed in XML: https://www.w3.org/TR/xml/#charsets @@ -876,7 +941,7 @@ def _sanitize(self, value): else: return value - def from_json(self, value): + def from_json(self, value) -> str | None: if value is None: return None elif isinstance(value, (bytes, str)): @@ -888,20 +953,23 @@ def from_json(self, value): else: raise TypeError('Value stored in a String must be None or a string, found %s' % type(value)) - def from_string(self, serialized): + def from_string(self, serialized) -> str | None: """String gets serialized and deserialized without quote marks.""" return self.from_json(serialized) - def to_string(self, value): + def to_string(self, value) -> str: """String gets serialized and deserialized without quote marks.""" if isinstance(value, bytes): value = value.decode('utf-8') return self.to_json(value) - @property - def none_to_xml(self): - """Returns True to use a XML node for the field and represent None as an attribute.""" - return True + def enforce_type(self, value: t.Any) -> str | None: + """ + (no-op override just to make mypy happy about XMLString.enforce_type) + """ + return super().enforce_type(value) + + none_to_xml = True # Use an XML node for the field, and represent None as an attribute. enforce_type = from_json @@ -915,7 +983,7 @@ class XMLString(String): an lxml.etree.XMLSyntaxError will be raised. """ - def to_json(self, value): + def to_json(self, value) -> t.Any: """ Serialize the data, ensuring that it is valid XML (or None). @@ -926,13 +994,13 @@ def to_json(self, value): value = self.enforce_type(value) return super().to_json(value) - def enforce_type(self, value): + def enforce_type(self, value: t.Any) -> str | None: if value is not None: etree.XML(value) - return value + return value # type: ignore -class DateTime(JSONField): +class DateTime(JSONField[t.Union[datetime, timedelta]]): """ A field for representing a datetime. @@ -942,7 +1010,7 @@ class DateTime(JSONField): DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%S.%f' - def from_json(self, value): + def from_json(self, value) -> datetime | timedelta | None: """ Parse the date from an ISO-formatted date string, or None. """ @@ -964,16 +1032,16 @@ def from_json(self, value): # Interpret raw numbers as a relative dates if isinstance(value, (int, float)): - value = datetime.timedelta(seconds=value) + value = timedelta(seconds=value) - if not isinstance(value, (datetime.datetime, datetime.timedelta)): + if not isinstance(value, (datetime, timedelta)): raise TypeError( "Value should be loaded from a string, a datetime object, a timedelta object, or None, not {}".format( type(value) ) ) - if isinstance(value, datetime.datetime): + if isinstance(value, datetime): if value.tzinfo is not None: return value.astimezone(pytz.utc) else: @@ -981,14 +1049,14 @@ def from_json(self, value): else: return value - def to_json(self, value): + def to_json(self, value) -> t.Any: """ Serialize the date as an ISO-formatted date string, or None. """ - if isinstance(value, datetime.datetime): + if isinstance(value, datetime): return value.strftime(self.DATETIME_FORMAT) - if isinstance(value, datetime.timedelta): + if isinstance(value, timedelta): return value.total_seconds() if value is None: @@ -996,14 +1064,14 @@ def to_json(self, value): raise TypeError(f"Value stored must be a datetime or timedelta object, not {type(value)}") - def to_string(self, value): + def to_string(self, value) -> str: """DateTime fields get serialized without quote marks.""" return self.to_json(value) enforce_type = from_json -class Any(JSONField): +class Any(JSONField[t.Any]): """ A field class for representing any piece of data; type is not enforced. @@ -1013,7 +1081,7 @@ class Any(JSONField): """ -class Reference(JSONField): +class Reference(JSONField[UsageKey]): """ An xblock reference. That is, a pointer to another xblock. @@ -1022,7 +1090,7 @@ class Reference(JSONField): """ -class ReferenceList(List): +class ReferenceList(List[UsageKey]): """ An list of xblock references. That is, pointers to xblocks. @@ -1033,7 +1101,19 @@ class ReferenceList(List): # but since Reference doesn't stipulate a definition for from/to, that seems unnecessary at this time. -class ReferenceValueDict(Dict): +class ReferenceListNotNone(ReferenceList): + """ + An list of xblock references. Should not equal None. + + Functionally, this is exactly equivalent to ReferenceList. + To the type-checker, this adds that guarantee that accessing the field will always return + a list of UsageKeys, rather than None OR a list of UsageKeys. + """ + def __get__(self, xblock: XBlock, xblock_class: type[XBlock]) -> list[UsageKey]: + return super().__get__(xblock, xblock_class) # type: ignore + + +class ReferenceValueDict(Dict[UsageKey]): """ A dictionary where the values are xblock references. That is, pointers to xblocks. @@ -1044,8 +1124,9 @@ class ReferenceValueDict(Dict): # but since Reference doesn't stipulate a definition for from/to, that seems unnecessary at this time. -def scope_key(instance, xblock): - """Generate a unique key for a scope that can be used as a +def scope_key(instance: Field, xblock: XBlock) -> str: + """ + Generate a unique key for a scope that can be used as a filename, in a URL, or in a KVS. Our goal is to have a pretty, human-readable 1:1 encoding. @@ -1110,11 +1191,10 @@ def scope_key(instance, xblock): key_list = [] - def encode(char): + def encode(char: str) -> str: """ Replace all non-alphanumeric characters with -n- where n is their Unicode codepoint. - TODO: Test for UTF8 which is not ASCII """ if char.isalnum(): return char diff --git a/xblock/internal.py b/xblock/internal.py index 12632aff0..62f725bc5 100644 --- a/xblock/internal.py +++ b/xblock/internal.py @@ -1,8 +1,11 @@ """ Internal machinery used to make building XBlock family base classes easier. """ +from __future__ import annotations + import functools import inspect +import typing as t class LazyClassProperty: @@ -13,9 +16,9 @@ class LazyClassProperty: executing the decorated method once, and then storing the result in the class __dict__. """ - def __init__(self, constructor): - self.__constructor = constructor - self.__cache = {} + def __init__(self, constructor: t.Callable): + self.__constructor: t.Callable = constructor + self.__cache: dict[object, object] = {} functools.wraps(self.__constructor)(self) def __get__(self, instance, owner): @@ -35,7 +38,12 @@ class NamedAttributesMetaclass(type): A metaclass which adds the __name__ attribute to all Nameable attributes which are attributes of the instantiated class, or of its baseclasses. """ - def __new__(mcs, name, bases, attrs): + def __new__( + mcs: type[NamedAttributesMetaclass], + name: str, + bases: tuple[type, ...], + attrs: dict[str, t.Any], + ): # Iterate over the attrs before they're bound to the class # so that we don't accidentally trigger any __get__ methods for attr_name, attr in attrs.items(): @@ -58,10 +66,10 @@ class Nameable: :class:`.NamedAttributesMetaclass`, will be assigned a `__name__` attribute based on what class attribute they are bound to. """ - __name__ = None + __name__: str | None = None @staticmethod - def needs_name(obj): + def needs_name(obj: object) -> bool: """ Return True if `obj` is a :class:`.Nameable` object that hasn't yet been assigned a name. diff --git a/xblock/mixins.py b/xblock/mixins.py index e921caf80..774562aea 100644 --- a/xblock/mixins.py +++ b/xblock/mixins.py @@ -2,6 +2,8 @@ This module defines all of the Mixins that provide components of XBlock-family functionality, such as ScopeStorage, RuntimeServices, and Handlers. """ +from __future__ import annotations + from collections import OrderedDict import copy import functools @@ -9,15 +11,23 @@ import logging import warnings import json +import typing as t +from opaque_keys.edx.keys import UsageKey from lxml import etree -from webob import Response +from webob import Request, Response from xblock.exceptions import JsonHandlerError, KeyValueMultiSaveError, XBlockSaveError, FieldDataDeprecationWarning -from xblock.fields import Field, Reference, Scope, ReferenceList +from xblock.fields import Field, Reference, Scope, ScopeIds, ReferenceListNotNone +from xblock.field_data import FieldData from xblock.internal import class_lazy, NamedAttributesMetaclass +if t.TYPE_CHECKING: + from xblock.core import XBlock + from xblock.runtime import Runtime + + # OrderedDict is used so that namespace attributes are put in predictable order # This allows for simple string equality assertions in tests and have no other effects XML_NAMESPACES = OrderedDict([ @@ -32,7 +42,7 @@ class HandlersMixin: """ @classmethod - def json_handler(cls, func): + def json_handler(cls, func: t.Callable[[XBlock, t.Any, str], t.Any]) -> t.Callable[[XBlock, Request, str], Response]: """ Wrap a handler to consume and produce JSON. @@ -51,7 +61,7 @@ def json_handler(cls, func): """ @cls.handler @functools.wraps(func) - def wrapper(self, request, suffix=''): + def wrapper(self, request: Request, suffix: str = '') -> Response: """The wrapper function `json_handler` returns.""" if request.method != "POST": return JsonHandlerError(405, "Method must be POST").get_response(allow=["POST"]) @@ -70,18 +80,19 @@ def wrapper(self, request, suffix=''): return wrapper @classmethod - def handler(cls, func): + def handler(cls, func: t.Callable) -> t.Callable: """ A decorator to indicate a function is usable as a handler. The wrapped function must return a `webob.Response` object. """ - func._is_xblock_handler = True # pylint: disable=protected-access + func._is_xblock_handler = True # type: ignore return func - def handle(self, handler_name, request, suffix=''): + def handle(self, handler_name: str, request: Request, suffix: str = '') -> Response: """Handle `request` with this block's runtime.""" - return self.runtime.handle(self, handler_name, request, suffix) + runtime: Runtime = self.runtime # type: ignore + return runtime.handle(self, handler_name, request, suffix) class RuntimeServicesMixin: @@ -188,7 +199,7 @@ def fields(cls): # pylint: disable=no-self-argument return fields - def __init__(self, scope_ids, field_data=None, **kwargs): + def __init__(self, scope_ids: ScopeIds, field_data: FieldData | None = None, **kwargs): """ Arguments: field_data (:class:`.FieldData`): Interface used by the XBlock @@ -212,9 +223,9 @@ def __init__(self, scope_ids, field_data=None, **kwargs): else: self._deprecated_per_instance_field_data = None # pylint: disable=invalid-name - self._field_data_cache = {} - self._dirty_fields = {} - self.scope_ids = scope_ids + self._field_data_cache: dict[str, t.Any] = {} + self._dirty_fields: dict[Field, t.Any] = {} + self.scope_ids: ScopeIds = scope_ids super().__init__(**kwargs) @@ -328,15 +339,19 @@ def __repr__(self): ) -class ChildrenModelMetaclass(ScopedStorageMixin.__class__): +class ChildrenModelMetaclass(NamedAttributesMetaclass): """ A metaclass that transforms the attribute `has_children = True` into a List field with a children scope. - """ - def __new__(mcs, name, bases, attrs): + def __new__( + mcs: type[ChildrenModelMetaclass], + name: str, + bases: tuple[type, ...], + attrs: dict[str, t.Any], + ): if (attrs.get('has_children', False) or any(getattr(base, 'has_children', False) for base in bases)): - attrs['children'] = ReferenceList( + attrs['children'] = ReferenceListNotNone( help='The ids of the children of this XBlock', scope=Scope.children) else: @@ -348,17 +363,23 @@ def __new__(mcs, name, bases, attrs): class HierarchyMixin(ScopedStorageMixin, metaclass=ChildrenModelMetaclass): """ This adds Fields for parents and children. + + TODO: Why in the world is this separate from XBlock?? + It depends on a bunch of XBlock attributes (runtime, has_childen, children...) + so it's not really useful outside of XBlock. As a future refactoring, consider + smashing this into the XBlock class and removing all the '_as_xblock' stuff below. """ parent = Reference(help='The id of the parent of this XBlock', default=None, scope=Scope.parent) + children: ReferenceListNotNone - def __init__(self, **kwargs): + def __init__(self, **kwargs) -> None: # A cache of the parent block, retrieved from .parent - self._parent_block = None - self._parent_block_id = None - self._child_cache = {} + self._parent_block: XBlock | None = None + self._parent_block_id: UsageKey | None = None + self._child_cache: dict[UsageKey, XBlock] = {} - for_parent = kwargs.pop('for_parent', None) + for_parent: XBlock | None = kwargs.pop('for_parent', None) # type: ignore if for_parent is not None: self._parent_block = for_parent @@ -366,57 +387,67 @@ def __init__(self, **kwargs): super().__init__(**kwargs) - def get_parent(self): + @property + def _as_xblock(self) -> XBlock: + """ + The same as 'self', but type-checks as an XBlock. + + This is a dumb mypy hack that would be great to remove. See TODO comment in class docstring. + """ + return self # type: ignore + + def get_parent(self) -> XBlock | None: """Return the parent block of this block, or None if there isn't one.""" if not self.has_cached_parent: - if self.parent is not None: - self._parent_block = self.runtime.get_block(self.parent) + if self._as_xblock.parent is not None: + self._parent_block = self._as_xblock.runtime.get_block(self._as_xblock.parent) else: self._parent_block = None - self._parent_block_id = self.parent + self._parent_block_id = self._as_xblock.parent return self._parent_block @property - def has_cached_parent(self): + def has_cached_parent(self) -> bool: """Return whether this block has a cached parent block.""" - return self.parent is not None and self._parent_block_id == self.parent + return self._as_xblock.parent is not None and self._parent_block_id == self._as_xblock.parent - def get_child(self, usage_id): + def get_child(self, usage_id: UsageKey) -> XBlock: """Return the child identified by ``usage_id``.""" if usage_id in self._child_cache: return self._child_cache[usage_id] - child_block = self.runtime.get_block(usage_id, for_parent=self) + runtime: Runtime = self.runtime # type: ignore + child_block = runtime.get_block(usage_id, for_parent=self) self._child_cache[usage_id] = child_block return child_block - def get_children(self, usage_id_filter=None): + def get_children(self, usage_id_filter: t.Callable[[UsageKey], bool] | None = None) -> list[XBlock]: """ Return instantiated XBlocks for each of this blocks ``children``. """ - if not self.has_children: + if not self._as_xblock.has_children: return [] return [ self.get_child(usage_id) - for usage_id in self.children + for usage_id in self._as_xblock.children if usage_id_filter is None or usage_id_filter(usage_id) ] - def clear_child_cache(self): + def clear_child_cache(self) -> None: """ Reset the cache of children stored on this XBlock. """ self._child_cache.clear() - def add_children_to_node(self, node): + def add_children_to_node(self, node: etree._Element) -> None: """ Add children to etree.Element `node`. """ - if self.has_children: - for child_id in self.children: - child = self.runtime.get_block(child_id) - self.runtime.add_block_as_child_node(child, node) + if self._as_xblock.has_children: + for child_id in self._as_xblock.children: + child = self._as_xblock.runtime.get_block(child_id) + self._as_xblock.runtime.add_block_as_child_node(child, node) class XmlSerializationMixin(ScopedStorageMixin): diff --git a/xblock/plugin.py b/xblock/plugin.py index 42f1ca6ce..e72f2d253 100644 --- a/xblock/plugin.py +++ b/xblock/plugin.py @@ -3,16 +3,20 @@ This code is in the Runtime layer. """ +from __future__ import annotations + import functools import itertools import logging -import pkg_resources +import typing as t + +from pkg_resources import iter_entry_points, EntryPoint from xblock.internal import class_lazy log = logging.getLogger(__name__) -PLUGIN_CACHE = {} +PLUGIN_CACHE: dict[tuple[str, str], type[Plugin]] = {} class PluginMissingError(Exception): @@ -21,35 +25,38 @@ class PluginMissingError(Exception): class AmbiguousPluginError(Exception): """Raised when a class name produces more than one entry_point.""" - def __init__(self, all_entry_points): + def __init__(self, all_entry_points: list[EntryPoint]): classes = (entpt.load() for entpt in all_entry_points) desc = ", ".join("{0.__module__}.{0.__name__}".format(cls) for cls in classes) msg = f"Ambiguous entry points for {all_entry_points[0].name}: {desc}" super().__init__(msg) -def default_select(identifier, all_entry_points): # pylint: disable=inconsistent-return-statements +def default_select(identifier: str, all_entry_points: list[EntryPoint]) -> EntryPoint: """ - Raise an exception when we have ambiguous entry points. + Raise an exception when we have no entry points or ambiguous entry points. """ - - if len(all_entry_points) == 0: + if not all_entry_points: raise PluginMissingError(identifier) - if len(all_entry_points) == 1: - return all_entry_points[0] elif len(all_entry_points) > 1: raise AmbiguousPluginError(all_entry_points) + return all_entry_points[0] class Plugin: - """Base class for a system that uses entry_points to load plugins. + """ + Base class for a system that uses entry_points to load plugins. Implementing classes are expected to have the following attributes: `entry_point`: The name of the entry point to load plugins from. - """ - entry_point = None # Should be overwritten by children classes + # TODO: In Python 3.11+, replace all annotations of `type[Plugin]` with + # `type[t.Self]`. The former is correct, but the latter is even better, + # because it'd mean that `XBlock.load_class` returns `type[XBlock]` instead + # of `type[Plugin]`. + + entry_point: str # Should be overwritten by children classes @class_lazy def extra_entry_points(cls): # pylint: disable=no-self-argument @@ -62,7 +69,7 @@ def extra_entry_points(cls): # pylint: disable=no-self-argument return [] @classmethod - def _load_class_entry_point(cls, entry_point): + def _load_class_entry_point(cls, entry_point: EntryPoint) -> type[Plugin]: """ Load `entry_point`, and set the `entry_point.name` as the attribute `plugin_name` on the loaded object @@ -72,7 +79,12 @@ def _load_class_entry_point(cls, entry_point): return class_ @classmethod - def load_class(cls, identifier, default=None, select=None): + def load_class( + cls, + identifier: str, + default: type[Plugin] | None = None, + select: t.Callable[[str, list[EntryPoint]], EntryPoint] | None = None, + ) -> type[Plugin]: """Load a single class specified by identifier. If `identifier` specifies more than a single class, and `select` is not None, @@ -100,7 +112,7 @@ def select(identifier, all_entry_points): if select is None: select = default_select - all_entry_points = list(pkg_resources.iter_entry_points(cls.entry_point, name=identifier)) + all_entry_points = list(iter_entry_points(cls.entry_point, name=identifier)) for extra_identifier, extra_entry_point in iter(cls.extra_entry_points): if identifier == extra_identifier: all_entry_points.append(extra_entry_point) @@ -117,7 +129,7 @@ def select(identifier, all_entry_points): return PLUGIN_CACHE[key] @classmethod - def load_classes(cls, fail_silently=True): + def load_classes(cls, fail_silently: bool = True) -> t.Iterable[tuple[str, type[Plugin]]]: """Load all the classes for a plugin. Produces a sequence containing the identifiers and their corresponding @@ -133,7 +145,7 @@ def load_classes(cls, fail_silently=True): contexts. Hence, the flag. """ all_classes = itertools.chain( - pkg_resources.iter_entry_points(cls.entry_point), + iter_entry_points(cls.entry_point), (entry_point for identifier, entry_point in iter(cls.extra_entry_points)), ) for class_ in all_classes: @@ -146,15 +158,15 @@ def load_classes(cls, fail_silently=True): raise @classmethod - def register_temp_plugin(cls, class_, identifier=None, dist='xblock'): - """Decorate a function to run with a temporary plugin available. + def register_temp_plugin(cls, class_: type, identifier: str | None = None, dist: str = 'xblock'): + """ + Decorate a function to run with a temporary plugin available. Use it like this in tests:: @register_temp_plugin(MyXBlockClass): def test_the_thing(): # Here I can load MyXBlockClass by name. - """ from unittest.mock import Mock # pylint: disable=import-outside-toplevel diff --git a/xblock/runtime.py b/xblock/runtime.py index 05e8af10a..23545dce3 100644 --- a/xblock/runtime.py +++ b/xblock/runtime.py @@ -1,6 +1,8 @@ """ Machinery to make the common case easy when building new runtimes """ +from __future__ import annotations + from abc import ABCMeta, abstractmethod from collections import namedtuple import functools @@ -12,11 +14,13 @@ import logging import re import threading +import typing as t import warnings from lxml import etree import markupsafe +from opaque_keys.edx.keys import DefinitionKey, UsageKey from web_fragments.fragment import Fragment from xblock.core import XBlock, XBlockAside, XML_NAMESPACES @@ -38,40 +42,47 @@ class KeyValueStore(metaclass=ABCMeta): """The abstract interface for Key Value Stores.""" - class Key(namedtuple("Key", "scope, user_id, block_scope_id, field_name, block_family")): + class Key(t.NamedTuple): """ Keys are structured to retain information about the scope of the data. Stores can use this information however they like to store and retrieve data. """ + scope: Scope + user_id: int | str | None + block_scope_id: UsageKey | DefinitionKey | str | None + field_name: str + block_family: str - def __new__(cls, scope, user_id, block_scope_id, field_name, block_family='xblock.v1'): + def __new__( # type: ignore + cls, scope, user_id, block_scope_id, field_name, block_family='xblock.v1' + ): return super(KeyValueStore.Key, cls).__new__(cls, scope, user_id, block_scope_id, field_name, block_family) @abstractmethod - def get(self, key): + def get(self, key: Key) -> t.Any: """Reads the value of the given `key` from storage.""" @abstractmethod - def set(self, key, value): + def set(self, key: Key, value: t.Any) -> None: """Sets `key` equal to `value` in storage.""" @abstractmethod - def delete(self, key): + def delete(self, key: Key) -> None: """Deletes `key` from storage.""" @abstractmethod - def has(self, key): + def has(self, key: Key) -> bool: """Returns whether or not `key` is present in storage.""" - def default(self, key): + def default(self, key: Key): """ Returns the context relevant default of the given `key` or raise KeyError which will result in the field's global default. """ raise KeyError(repr(key)) - def set_many(self, update_dict): + def set_many(self, update_dict: dict[Key, t.Any]) -> None: """ For each (`key, value`) in `update_dict`, set `key` to `value` in storage. @@ -90,8 +101,8 @@ class DictKeyValueStore(KeyValueStore): A `KeyValueStore` that stores everything into a Python dictionary. """ - def __init__(self, storage=None): - self.db_dict = storage if storage is not None else {} + def __init__(self, storage: dict[KeyValueStore.Key, t.Any] | None = None): + self.db_dict: dict[KeyValueStore.Key, t.Any] = storage if storage is not None else {} def get(self, key): return self.db_dict[key] @@ -115,14 +126,14 @@ class KvsFieldData(FieldData): that uses the correct scoped keys for the underlying KeyValueStore """ - def __init__(self, kvs, **kwargs): + def __init__(self, kvs: KeyValueStore, **kwargs): super().__init__(**kwargs) self._kvs = kvs - def __repr__(self): + def __repr__(self) -> str: return "{0.__class__.__name__}({0._kvs!r})".format(self) - def _getfield(self, block, name): + def _getfield(self, block: XBlock, name: str) -> Field: """ Return the field with the given `name` from `block`. If no field with `name` exists in any namespace, raises a KeyError. @@ -143,7 +154,7 @@ def _getfield(self, block, name): # really doesn't name a field raise KeyError(name) - def _key(self, block, name): + def _key(self, block: XBlock, name: str) -> KeyValueStore.Key: """ Resolves `name` to a key, in the following form: @@ -156,6 +167,7 @@ def _key(self, block, name): ) """ field = self._getfield(block, name) + block_id: UsageKey | DefinitionKey | str | None # Type depends on field scope. if field.scope in (Scope.children, Scope.parent): block_id = block.scope_ids.usage_id user_id = None @@ -185,7 +197,7 @@ def _key(self, block, name): ) return key - def get(self, block, name): + def get(self, block: XBlock, name: str) -> t.Any: """ Retrieve the value for the field named `name`. @@ -194,19 +206,19 @@ def get(self, block, name): """ return self._kvs.get(self._key(block, name)) - def set(self, block, name, value): + def set(self, block: XBlock, name: str, value: t.Any) -> None: """ Set the value of the field named `name` """ self._kvs.set(self._key(block, name), value) - def delete(self, block, name): + def delete(self, block: XBlock, name: str) -> None: """ Reset the value of the field named `name` to the default """ self._kvs.delete(self._key(block, name)) - def has(self, block, name): + def has(self, block: XBlock, name: str) -> bool: """ Return whether or not the field named `name` has a non-default value """ @@ -215,7 +227,7 @@ def has(self, block, name): except KeyError: return False - def set_many(self, block, update_dict): + def set_many(self, block, update_dict: dict[str, t.Any]): """Update the underlying model with the correct values.""" updated_dict = {} @@ -225,7 +237,7 @@ def set_many(self, block, update_dict): self._kvs.set_many(updated_dict) - def default(self, block, name): + def default(self, block: XBlock, name: str) -> t.Any: """ Ask the kvs for the default (default implementation which other classes may override). @@ -360,8 +372,8 @@ def create_definition(self, block_type, slug=None): class MemoryIdManager(IdReader, IdGenerator): """A simple dict-based implementation of IdReader and IdGenerator.""" - ASIDE_USAGE_ID = namedtuple('MemoryAsideUsageId', 'usage_id aside_type') - ASIDE_DEFINITION_ID = namedtuple('MemoryAsideDefinitionId', 'definition_id aside_type') + ASIDE_USAGE_ID = namedtuple('MemoryAsideUsageId', 'usage_id aside_type') # type: ignore[name-match] + ASIDE_DEFINITION_ID = namedtuple('MemoryAsideDefinitionId', 'definition_id aside_type') # type: ignore[name-match] def __init__(self): self._ids = itertools.count() @@ -1246,7 +1258,7 @@ def __delattr__(self, name): # Cache of Mixologist generated classes -_CLASS_CACHE = {} +_CLASS_CACHE: dict[tuple[type, tuple[type, ...]], type] = {} _CLASS_CACHE_LOCK = threading.RLock()