diff --git a/src/viur/core/__init__.py b/src/viur/core/__init__.py index 2e8883f8d..363932aca 100644 --- a/src/viur/core/__init__.py +++ b/src/viur/core/__init__.py @@ -34,7 +34,7 @@ import inspect import warnings from types import ModuleType -from typing import Callable, Dict, Union, List +import typing as t from google.appengine.api import wrap_wsgi_app import yaml from viur.core import i18n, request, utils @@ -84,7 +84,8 @@ # Show DeprecationWarning from the viur-core warnings.filterwarnings("always", category=DeprecationWarning, module=r"viur\.core.*") -def load_indexes_from_file() -> Dict[str, List]: + +def load_indexes_from_file() -> dict[str, list]: """ Loads all indexes from the index.yaml and stores it in a dictionary sorted by the module(kind) :return A dictionary of indexes per module @@ -127,7 +128,7 @@ def setDefaultDomainLanguage(domain: str, lang: str): conf.i18n.domain_language_mapping[host] = lang.lower() -def buildApp(modules: Union[ModuleType, object], renderers: Union[ModuleType, Dict], default: str = None) -> Module: +def buildApp(modules: ModuleType | object, renderers: ModuleType | object, default: str = None) -> Module: """ Creates the application-context for the current instance. @@ -234,7 +235,7 @@ def buildApp(modules: Union[ModuleType, object], renderers: Union[ModuleType, Di return root -def setup(modules: Union[object, ModuleType], render: Union[ModuleType, Dict] = None, default: str = "html"): +def setup(modules: ModuleType | object, render: ModuleType | object = None, default: str = "html"): """ Define whats going to be served by this instance. @@ -310,7 +311,7 @@ def setup(modules: Union[object, ModuleType], render: Union[ModuleType, Dict] = return wrap_wsgi_app(app) -def app(environ: dict, start_response: Callable): +def app(environ: dict, start_response: t.Callable): return request.Router(environ).response(environ, start_response) diff --git a/src/viur/core/bones/base.py b/src/viur/core/bones/base.py index 45a67df43..b44f23b2b 100644 --- a/src/viur/core/bones/base.py +++ b/src/viur/core/bones/base.py @@ -12,7 +12,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum -from typing import Any, Dict, Iterator, List, Optional, Set, Tuple, Union +import typing as t from viur.core import db, utils from viur.core.config import conf @@ -76,9 +76,9 @@ class ReadFromClientError: """A ReadFromClientErrorSeverity enumeration value representing the severity of the error.""" errorMessage: str """A string containing a human-readable error message describing the issue.""" - fieldPath: List[str] = field(default_factory=list) + fieldPath: list[str] = field(default_factory=list) """A list of strings representing the path to the field where the error occurred.""" - invalidatedFields: List[str] = None + invalidatedFields: list[str] = None """A list of strings containing the names of invalidated fields, if any.""" @@ -192,18 +192,18 @@ def __init__( self, *, compute: Compute = None, - defaultValue: Any = None, + defaultValue: t.Any = None, descr: str = "", getEmptyValueFunc: callable = None, indexed: bool = True, isEmptyFunc: callable = None, # fixme: Rename this, see below. - languages: Union[None, List[str]] = None, - multiple: Union[bool, MultipleConstraints] = False, - params: Dict = None, + languages: None | list[str] = None, + multiple: bool | MultipleConstraints = False, + params: dict = None, readOnly: bool = None, # fixme: Rename into readonly (all lowercase!) soon. - required: Union[bool, List[str], Tuple[str]] = False, + required: bool | list[str] | tuple[str] = False, searchable: bool = False, - unique: Union[None, UniqueValue] = None, + unique: None | UniqueValue = None, vfunc: callable = None, # fixme: Rename this, see below. visible: bool = True, ): @@ -306,7 +306,7 @@ def isInvalid(self, value): """ return False - def isEmpty(self, value: Any) -> bool: + def isEmpty(self, value: t.Any) -> bool: """ Check if the given single value represents the "empty" value. This usually is the empty string, 0 or False. @@ -339,7 +339,7 @@ def getDefaultValue(self, skeletonInstance): else: return self.defaultValue - def getEmptyValue(self) -> Any: + def getEmptyValue(self) -> t.Any: """ Returns the value representing an empty field for this bone. This might be the empty string for str/text Bones, Zero for numeric bones etc. @@ -482,9 +482,9 @@ def parseSubfieldsFromClient(self) -> bool: """ return False - def singleValueFromClient(self, value: Any, skel: 'SkeletonInstance', + def singleValueFromClient(self, value: t.Any, skel: 'SkeletonInstance', bone_name: str, client_data: dict - ) -> tuple[Any, list[ReadFromClientError] | None]: + ) -> tuple[t.Any, list[ReadFromClientError] | None]: """Load a single value from a client :param value: The single value which should be loaded. @@ -501,7 +501,7 @@ def singleValueFromClient(self, value: Any, skel: 'SkeletonInstance', return self.getEmptyValue(), [ ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Will not read a BaseBone fromClient!")] - def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> Union[None, List[ReadFromClientError]]: + def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> None | list[ReadFromClientError]: """ Reads a value from the client and stores it in the skeleton instance if it is valid for the bone. @@ -586,7 +586,7 @@ def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> Union[N errors.extend(self.validateMultipleConstraints(skel, name)) return errors or None - def validateMultipleConstraints(self, skel: 'SkeletonInstance', name: str) -> List[ReadFromClientError]: + def validateMultipleConstraints(self, skel: 'SkeletonInstance', name: str) -> list[ReadFromClientError]: """ Validates the value of a bone against its multiple constraints and returns a list of ReadFromClientError objects for each violation, such as too many items or duplicates. @@ -844,8 +844,8 @@ def buildDBFilter(self, name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None) -> db.Query: + rawFilter: dict, + prefix: t.Optional[str] = None) -> db.Query: """ Parses the searchfilter a client specified in his Request into something understood by the datastore. @@ -896,7 +896,7 @@ def buildDBSort(self, name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict) -> Optional[db.Query]: + rawFilter: dict) -> t.Optional[db.Query]: """ Same as buildDBFilter, but this time its not about filtering the results, but by sorting them. @@ -945,7 +945,7 @@ def buildDBSort(self, dbFilter.order(order) return dbFilter - def _hashValueForUniquePropertyIndex(self, value: Union[str, int]) -> List[str]: + def _hashValueForUniquePropertyIndex(self, value: str | int) -> list[str]: """ Generates a hash of the given value for creating unique property indexes. @@ -958,7 +958,7 @@ def _hashValueForUniquePropertyIndex(self, value: Union[str, int]) -> List[str]: :return: A list containing a string representation of the hashed value. If the bone is multiple, the list may contain more than one hashed value. """ - def hashValue(value: Union[str, int]) -> str: + def hashValue(value: str | int) -> str: h = hashlib.sha256() h.update(str(value).encode("UTF-8")) res = h.hexdigest() @@ -994,7 +994,7 @@ def keyHash(key): # Lock the value for that specific list return [hashValue(", ".join(tmpList))] - def getUniquePropertyIndexValues(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> List[str]: + def getUniquePropertyIndexValues(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> list[str]: """ Returns a list of hashes for the current value(s) of a bone in the skeleton, used for storing in the unique property value index. @@ -1011,13 +1011,13 @@ def getUniquePropertyIndexValues(self, skel: 'viur.core.skeleton.SkeletonInstanc return [] return self._hashValueForUniquePropertyIndex(val) - def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Returns a set of blob keys referenced from this bone """ return set() - def performMagic(self, valuesCache: Dict, name: str, isAdd: bool): + def performMagic(self, valuesCache: dict, name: str, isAdd: bool): """ This function applies "magically" functionality which f.e. inserts the current Date or the current user. @@ -1051,7 +1051,7 @@ def refresh(self, skel: 'viur.core.skeleton.SkeletonInstance', boneName: str) -> """ pass - def mergeFrom(self, valuesCache: Dict, boneName: str, otherSkel: 'viur.core.skeleton.SkeletonInstance'): + def mergeFrom(self, valuesCache: dict, boneName: str, otherSkel: 'viur.core.skeleton.SkeletonInstance'): """ Merges the values from another skeleton instance into the current instance, given that the bone types match. @@ -1076,9 +1076,9 @@ def mergeFrom(self, valuesCache: Dict, boneName: str, otherSkel: 'viur.core.skel def setBoneValue(self, skel: 'SkeletonInstance', boneName: str, - value: Any, + value: t.Any, append: bool, - language: Union[None, str] = None) -> bool: + language: None | str = None) -> bool: """ Sets the value of a bone in a skeleton instance, with optional support for appending and language-specific values. Sanity checks are being performed. @@ -1132,7 +1132,7 @@ def setBoneValue(self, skel[boneName][language] = val return True - def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Returns a set of strings as search index for this bone. @@ -1150,7 +1150,7 @@ def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) def iter_bone_value( self, skel: 'viur.core.skeleton.SkeletonInstance', name: str - ) -> Iterator[Tuple[Optional[int], Optional[str], Any]]: + ) -> t.Iterator[tuple[t.Optional[int], t.Optional[str], t.Any]]: """ Yield all values from the Skeleton related to this bone instance. diff --git a/src/viur/core/bones/boolean.py b/src/viur/core/bones/boolean.py index 9c57439cb..c16270539 100644 --- a/src/viur/core/bones/boolean.py +++ b/src/viur/core/bones/boolean.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Union +import typing as t from viur.core import conf, db, utils from viur.core.bones.base import BaseBone @@ -18,11 +18,7 @@ class BooleanBone(BaseBone): def __init__( self, *, - defaultValue: Union[ - bool, - List[bool], - Dict[str, Union[List[bool], bool]], - ] = None, + defaultValue: bool | list[bool] | dict[str, list[bool] | bool] = None, **kwargs ): if defaultValue is None: @@ -56,7 +52,7 @@ def getEmptyValue(self): """ return False - def isEmpty(self, value: Any): + def isEmpty(self, value: t.Any): """ Checks if the given boolean value is empty. @@ -85,8 +81,8 @@ def buildDBFilter( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None + rawFilter: dict, + prefix: t.Optional[str] = None ) -> db.Query: """ Builds a database filter based on the boolean value. diff --git a/src/viur/core/bones/captcha.py b/src/viur/core/bones/captcha.py index 024bda88a..b0e1ff8e1 100644 --- a/src/viur/core/bones/captcha.py +++ b/src/viur/core/bones/captcha.py @@ -1,7 +1,7 @@ import json import urllib.parse import urllib.request -from typing import List, Union +import typing as t from viur.core import conf, current from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity @@ -49,7 +49,7 @@ def unserialize(self, skel, name) -> bool: skel.accessedValues[name] = self.publicKey return True - def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> Union[None, List[ReadFromClientError]]: + def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> None | list[ReadFromClientError]: """ Reads a value from the client. If this value is valid for this bone, diff --git a/src/viur/core/bones/date.py b/src/viur/core/bones/date.py index bcd21d05b..44646947e 100644 --- a/src/viur/core/bones/date.py +++ b/src/viur/core/bones/date.py @@ -1,5 +1,5 @@ from datetime import datetime, timedelta, timezone -from typing import Dict, Optional +import typing as t import pytz import tzlocal @@ -316,8 +316,8 @@ def buildDBFilter(self, name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None) -> db.Query: + rawFilter: dict, + prefix: t.Optional[str] = None) -> db.Query: """ Constructs a datastore filter for date and/or time values based on the given raw filter. It parses the raw filter and, if successful, applies it to the datastore query. diff --git a/src/viur/core/bones/file.py b/src/viur/core/bones/file.py index 9dc51eedf..1769a07b8 100644 --- a/src/viur/core/bones/file.py +++ b/src/viur/core/bones/file.py @@ -7,7 +7,7 @@ from hashlib import sha256 from time import time -from typing import Any, Dict, List, Set, Union +import typing as t from viur.core import conf, db from viur.core.bones.treeleaf import TreeLeafBone from viur.core.tasks import CallDeferred @@ -16,7 +16,7 @@ @CallDeferred -def ensureDerived(key: db.Key, srcKey, deriveMap: Dict[str, Any], refreshKey: db.Key = None): +def ensureDerived(key: db.Key, srcKey, deriveMap: dict[str, t.Any], refreshKey: db.Key = None): r""" The function is a deferred function that ensures all pending thumbnails or other derived files are built. It takes the following parameters: @@ -24,7 +24,7 @@ def ensureDerived(key: db.Key, srcKey, deriveMap: Dict[str, Any], refreshKey: db :param db.key key: The database key of the file-object that needs to have its derivation map updated. :param str srcKey: A prefix for a stable key to prevent rebuilding derived files repeatedly. - :param Dict[str,Any] deriveMap: A list of DeriveDicts that need to be built or updated. + :param dict[str,Any] deriveMap: A list of DeriveDicts that need to be built or updated. :param db.Key refreshKey: If set, the function fetches and refreshes the skeleton after building new derived files. @@ -139,9 +139,9 @@ class FileBone(TreeLeafBone): def __init__( self, *, - derive: Union[None, Dict[str, Any]] = None, - maxFileSize: Union[None, int] = None, - validMimeTypes: Union[None, List[str]] = None, + derive: None | dict[str, t.Any] = None, + maxFileSize: None | int = None, + validMimeTypes: None | list[str] = None, **kwargs ): r""" @@ -231,7 +231,7 @@ def handleDerives(values): else: handleDerives(values) - def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: r""" Retrieves the referenced blobs in the FileBone. diff --git a/src/viur/core/bones/json.py b/src/viur/core/bones/json.py index ae47eaca2..4aeae0458 100644 --- a/src/viur/core/bones/json.py +++ b/src/viur/core/bones/json.py @@ -1,6 +1,6 @@ import ast import json -from typing import Mapping, Union +import typing as t import jsonschema @@ -23,7 +23,7 @@ class JsonBone(RawBone): type = "raw.json" - def __init__(self, indexed: bool = False, multiple: bool = False, languages: bool = None, schema: Mapping = {}, + def __init__(self, indexed: bool = False, multiple: bool = False, languages: bool = None, schema: t.Mapping = {}, *args, **kwargs): super().__init__(*args, **kwargs) @@ -52,7 +52,7 @@ def unserialize(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> return False - def singleValueFromClient(self, value: Union[str, list, dict], skel, bone_name, client_data): + def singleValueFromClient(self, value: str | list | dict, skel, bone_name, client_data): if value: if not isinstance(value, (list, dict)): value = str(value) diff --git a/src/viur/core/bones/key.py b/src/viur/core/bones/key.py index 07581a940..92c1e3ad2 100644 --- a/src/viur/core/bones/key.py +++ b/src/viur/core/bones/key.py @@ -1,6 +1,6 @@ import copy import logging -from typing import Dict, List, Optional, Union +import typing as t from viur.core import db, utils from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity @@ -27,7 +27,7 @@ def __init__( descr: str = "Key", readOnly: bool = True, # default is readonly visible: bool = False, # default is invisible - allowed_kinds: Union[None, List[str]] = None, # None allows for any kind + allowed_kinds: None | list[str] = None, # None allows for any kind check: bool = False, # check for entity existence **kwargs ): @@ -153,8 +153,8 @@ def buildDBFilter( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None + rawFilter: dict, + prefix: t.Optional[str] = None ) -> db.Query: """ This method parses the search filter specified by the client in their request and converts diff --git a/src/viur/core/bones/numeric.py b/src/viur/core/bones/numeric.py index 384298b37..1b54154ac 100644 --- a/src/viur/core/bones/numeric.py +++ b/src/viur/core/bones/numeric.py @@ -1,6 +1,6 @@ import logging import warnings -from typing import Any, Dict, Optional, Set, Union +import typing as t import sys @@ -29,8 +29,8 @@ class NumericBone(BaseBone): def __init__( self, *, - max: Union[int, float] = MAX, - min: Union[int, float] = MIN, + max: int | float = MAX, + min: int | float = MIN, mode=None, # deprecated! precision: int = 0, **kwargs @@ -98,7 +98,7 @@ def getEmptyValue(self): else: return 0 - def isEmpty(self, value: Any): + def isEmpty(self, value: t.Any): """ This method checks if a given raw value is considered empty for the NumericBone instance. It attempts to convert the raw value into a valid numeric value (integer or floating-point @@ -139,8 +139,8 @@ def buildDBFilter( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None + rawFilter: dict, + prefix: t.Optional[str] = None ) -> db.Query: updatedFilter = {} @@ -162,7 +162,7 @@ def buildDBFilter( return super().buildDBFilter(name, skel, dbFilter, updatedFilter, prefix) - def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ This method generates a set of search tags based on the numeric values stored in the NumericBone instance. It iterates through the bone values and adds the string representation of each value @@ -179,7 +179,7 @@ def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) result.add(str(value)) return result - def _convert_to_numeric(self, value: Any) -> int | float: + def _convert_to_numeric(self, value: t.Any) -> int | float: """Convert a value to an int or float considering the precision. If the value is not convertable an exception will be raised.""" @@ -199,7 +199,7 @@ def refresh(self, skel: 'viur.core.skeleton.SkeletonInstance', boneName: str) -> """ super().refresh(skel, boneName) - def refresh_single_value(value: Any) -> float | int: + def refresh_single_value(value: t.Any) -> float | int: if value == "": return self.getEmptyValue() elif not isinstance(value, (int, float, type(None))): diff --git a/src/viur/core/bones/password.py b/src/viur/core/bones/password.py index c361244a9..1aa9d876d 100644 --- a/src/viur/core/bones/password.py +++ b/src/viur/core/bones/password.py @@ -6,7 +6,6 @@ import hashlib import re -from typing import List, Tuple, Union from viur.core import conf, utils from viur.core.bones.string import StringBone @@ -66,7 +65,7 @@ def __init__( *, descr: str = "Password", test_threshold: int = 4, - tests: List[Tuple] = tests, + tests: list[tuple] = tests, **kwargs ): """ @@ -112,7 +111,7 @@ def isInvalid(self, value): return False - def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> Union[None, List[ReadFromClientError]]: + def fromClient(self, skel: 'SkeletonInstance', name: str, data: dict) -> None | list[ReadFromClientError]: """ Processes the password field from the client data, validates it, and stores it in the skeleton instance after hashing. This method performs several checks, such as ensuring that diff --git a/src/viur/core/bones/randomslice.py b/src/viur/core/bones/randomslice.py index 66ec5dbf7..9ecd8a268 100644 --- a/src/viur/core/bones/randomslice.py +++ b/src/viur/core/bones/randomslice.py @@ -1,5 +1,5 @@ from random import random, sample, shuffle -from typing import Dict, List, Optional +import typing as t from itertools import chain from math import ceil @@ -59,8 +59,8 @@ def buildDBSort( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict - ) -> Optional[db.Query]: + rawFilter: dict + ) -> t.Optional[db.Query]: """ Modifies the database query to return a random selection of elements by creating multiple subqueries, each covering a slice of the data. This method doesn't just change the order of @@ -138,7 +138,8 @@ def calculateInternalMultiQueryLimit(self, query: db.Query, targetAmount: int) - """ return ceil(targetAmount * self.sliceSize) - def customMultiQueryMerge(self, dbFilter: db.Query, result: List[db.Entity], targetAmount: int) -> List[db.Entity]: + def customMultiQueryMerge(self, dbFilter: db.Query, result: list[db.Entity], targetAmount: int) \ + -> list[db.Entity]: """ Merges the results of multiple subqueries by randomly selecting 'targetAmount' elements from the combined 'result' list. diff --git a/src/viur/core/bones/record.py b/src/viur/core/bones/record.py index 3367c15bf..0ce8800fa 100644 --- a/src/viur/core/bones/record.py +++ b/src/viur/core/bones/record.py @@ -1,4 +1,3 @@ -from typing import List, Set from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity @@ -98,7 +97,7 @@ def singleValueFromClient(self, value, skel, bone_name, client_data): ) return usingSkel, usingSkel.errors - def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Collects search tags from the 'using' skeleton instance for the given bone. @@ -147,7 +146,7 @@ def getValues(res, skel, valuesCache, searchPrefix): return res - def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Retrieves a set of referenced blobs for the given skeleton instance and name. @@ -170,7 +169,7 @@ def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: return result - def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> List[str]: + def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> list[str]: """ This method is intentionally not implemented as it's not possible to determine how to derive a key from the related skeleton being used (i.e., which fields to include and how). diff --git a/src/viur/core/bones/relational.py b/src/viur/core/bones/relational.py index c5a192f7e..be98a3501 100644 --- a/src/viur/core/bones/relational.py +++ b/src/viur/core/bones/relational.py @@ -5,7 +5,7 @@ import logging import warnings from enum import Enum -from typing import Any, Dict, List, Optional, Set, Union +import typing as t from itertools import chain from time import time @@ -167,11 +167,11 @@ def __init__( consistency: RelationalConsistency = RelationalConsistency.Ignore, format: str = "$(dest.name)", kind: str = None, - module: Optional[str] = None, - parentKeys: Optional[List[str]] = None, - refKeys: Optional[List[str]] = None, + module: t.Optional[str] = None, + parentKeys: t.Optional[list[str]] = None, + refKeys: t.Optional[list[str]] = None, updateLevel: RelationalUpdateLevel = RelationalUpdateLevel.Always, - using: Optional['viur.core.skeleton.RelSkel'] = None, + using: t.Optional['viur.core.skeleton.RelSkel'] = None, **kwargs ): """ @@ -796,8 +796,8 @@ def buildDBFilter( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None + rawFilter: dict, + prefix: t.Optional[str] = None ) -> db.Query: """ Builds a datastore query by modifying the given filter based on the RelationalBone's properties. @@ -902,8 +902,8 @@ def buildDBSort( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict - ) -> Optional[db.Query]: + rawFilter: dict + ) -> t.Optional[db.Query]: """ Builds a datastore query by modifying the given filter based on the RelationalBone's properties for sorting. @@ -916,7 +916,7 @@ def buildDBSort( :param dict rawFilter: The raw filter applied to the original datastore query. :return: The modified datastore query with updated sorting behavior. - :rtype: Optional[db.Query] + :rtype: t.Optional[db.Query] :raises RuntimeError: If the sorting is invalid, e.g., using properties not in 'refKeys' or not a bone in 'using'. @@ -1123,7 +1123,7 @@ def updateInplace(relDict): for k in skel[boneName]: updateInplace(k) - def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Retrieves the search tags for the given RelationalBone in the provided skeleton. @@ -1156,7 +1156,7 @@ def get_values(skel_, values_cache): return result - def createRelSkelFromKey(self, key: Union[str, db.Key], rel: Union[dict, None] = None): + def createRelSkelFromKey(self, key: t.Union[str, "db.Key"], rel: dict | None = None): """ Creates a relSkel instance valid for this bone from the given database key. @@ -1191,9 +1191,9 @@ def setBoneValue( self, skel: 'SkeletonInstance', boneName: str, - value: Any, + value: t.Any, append: bool, - language: Union[None, str] = None + language: None | str = None ) -> bool: """ Sets the value of the specified bone in the given skeleton. Sanity checks are performed to ensure the @@ -1282,7 +1282,7 @@ def setBoneValue( skel[boneName] = tmpRes return True - def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Retrieves the set of referenced blobs from the specified bone in the given skeleton instance. @@ -1304,7 +1304,7 @@ def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: result.update(bone_.getReferencedBlobs(value["rel"], key)) return result - def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> List[str]: + def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> list[str]: """ Generates unique property index values for the RelationalBone based on the referenced keys. Can be overridden if different behavior is required (e.g., examining values from `prop:usingSkel`). diff --git a/src/viur/core/bones/select.py b/src/viur/core/bones/select.py index 35c32f9c3..30f4dcbbf 100644 --- a/src/viur/core/bones/select.py +++ b/src/viur/core/bones/select.py @@ -1,7 +1,7 @@ import enum from collections import OrderedDict from numbers import Number -from typing import Any, Callable, Dict, List, TYPE_CHECKING, Tuple, Union +import typing as t from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity from viur.core.i18n import translate @@ -11,15 +11,15 @@ except ImportError: Self = BaseBone # SelectBone is not defined here and Self is not available -if TYPE_CHECKING: +if t.TYPE_CHECKING: from viur.core.skeleton import SkeletonInstance -SelectBoneValue = Union[str, Number, enum.Enum] +SelectBoneValue = t.Union[str, Number, enum.Enum] """ Type alias of possible values in a SelectBone. SelectBoneValue can be either a string (str) or a number (Number) """ -SelectBoneMultiple = List[SelectBoneValue] +SelectBoneMultiple = list[SelectBoneValue] """ SelectBoneMultiple is a list of SelectBoneValue elements.""" @@ -37,13 +37,13 @@ class SelectBone(BaseBone): def __init__( self, *, - defaultValue: Union[ + defaultValue: t.Union[ SelectBoneValue, SelectBoneMultiple, - Dict[str, Union[SelectBoneMultiple, SelectBoneValue]], - Callable[["SkeletonInstance", Self], Any], + t.Dict[str, t.Union[SelectBoneMultiple, SelectBoneValue]], + t.Callable[["SkeletonInstance", Self], t.Any], ] = None, - values: Union[Dict, List, Tuple, Callable, enum.EnumMeta] = (), + values: dict | list | tuple | t.Callable | enum.EnumMeta = (), **kwargs ): diff --git a/src/viur/core/bones/sortindex.py b/src/viur/core/bones/sortindex.py index 49ba6fa59..f85413d5e 100644 --- a/src/viur/core/bones/sortindex.py +++ b/src/viur/core/bones/sortindex.py @@ -1,7 +1,5 @@ -import typing - +import typing as t import time - from viur.core.bones.numeric import NumericBone @@ -10,7 +8,7 @@ class SortIndexBone(NumericBone): The SortIndexBone class is specifically designed to handle sorting indexes for data elements, which are numeric values that determine the order of these elements. It inherits from the NumericBone. - :param typing.Union[int, float] defaultValue: A default value for the bone, which is a function that returns + :param int | float defaultValue: A default value for the bone, which is a function that returns the current time by default. This parameter accepts either an integer or a floating-point number. :param str descr: A short description of the bone, set to "SortIndex" by default. :param int precision: The precision of the numeric value, determining the number of decimal places allowed. @@ -22,7 +20,7 @@ class SortIndexBone(NumericBone): def __init__( self, *, - defaultValue: typing.Union[int, float] = lambda *args, **kwargs: time.time(), + defaultValue: int | float = lambda *args, **kwargs: time.time(), descr: str = "SortIndex", precision: int = 8, **kwargs diff --git a/src/viur/core/bones/spatial.py b/src/viur/core/bones/spatial.py index 189ebd93c..50c03c9d4 100644 --- a/src/viur/core/bones/spatial.py +++ b/src/viur/core/bones/spatial.py @@ -6,7 +6,7 @@ import logging from copy import deepcopy -from typing import Any, Dict, List, Optional, Tuple, Union +import typing as t import math from math import floor @@ -70,8 +70,8 @@ class SpatialBone(BaseBone): type = "spatial" - def __init__(self, *, boundsLat: Tuple[float, float], boundsLng: Tuple[float, float], - gridDimensions: Tuple[int, int], **kwargs): + def __init__(self, *, boundsLat: tuple[float, float], boundsLng: tuple[float, float], + gridDimensions: tuple[int, int], **kwargs): """ Initializes a new SpatialBone. @@ -110,7 +110,7 @@ def getGridSize(self): lngDelta = float(self.boundsLng[1] - self.boundsLng[0]) return latDelta / float(self.gridDimensions[0]), lngDelta / float(self.gridDimensions[1]) - def isInvalid(self, value: Tuple[float, float]) -> Union[str, bool]: + def isInvalid(self, value: tuple[float, float]) -> str | bool: """ Validate if the given point (latitude, longitude) falls within the specified boundaries. Rejects all values outside the defined region. @@ -183,7 +183,7 @@ def parseSubfieldsFromClient(self): """ return True # We'll always get .lat and .lng - def isEmpty(self, value: Any): + def isEmpty(self, value: t.Any): """ Check if the given raw value is considered empty (either not present or equal to the empty value). @@ -202,7 +202,7 @@ def isEmpty(self, value: Any): return True return value == self.getEmptyValue() - def getEmptyValue(self) -> Tuple[float, float]: + def getEmptyValue(self) -> tuple[float, float]: """ Returns an empty value for the bone, which represents an invalid position. Use 91.0, 181.0 as a special marker for empty, as they are both out of range for Latitude (-90, +90) and Longitude (-180, 180), but will @@ -240,8 +240,8 @@ def buildDBFilter( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None + rawFilter: dict, + prefix: t.Optional[str] = None ) -> db.Query: """ Parses the client's search filter specified in their request and converts it into a format understood by the @@ -314,8 +314,8 @@ def calculateInternalMultiQueryLimit(self, dbQuery: db.Query, targetAmount: int) return targetAmount * 2 def customMultiQueryMerge(self, name, lat, lng, dbFilter: db.Query, - result: List[db.Entity], targetAmount: int - ) -> List[db.Entity]: + result: list[db.Entity], targetAmount: int + ) -> list[db.Entity]: """ Randomly returns 'targetAmount' elements from 'result'. @@ -367,9 +367,9 @@ def setBoneValue( self, skel: 'SkeletonInstance', boneName: str, - value: Any, + value: t.Any, append: bool, - language: Union[None, str] = None + language: None | str = None ) -> bool: """ Sets the value of the bone to the provided 'value'. diff --git a/src/viur/core/bones/string.py b/src/viur/core/bones/string.py index 76b7c5aa9..f3a01a759 100644 --- a/src/viur/core/bones/string.py +++ b/src/viur/core/bones/string.py @@ -1,7 +1,7 @@ import warnings import logging -from typing import Callable, Dict, List, Optional, Set +import typing as t from viur.core import current, db, utils from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity @@ -19,7 +19,7 @@ def __init__( caseSensitive: bool = True, max_length: int | None = 254, min_length: int | None = None, - natural_sorting: bool | Callable = False, + natural_sorting: bool | t.Callable = False, **kwargs ): """ @@ -142,8 +142,8 @@ def buildDBFilter( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict, - prefix: Optional[str] = None + rawFilter: dict, + prefix: t.Optional[str] = None ) -> db.Query: """ Builds and returns a database filter for this data field based on the provided raw filter data. @@ -210,8 +210,8 @@ def buildDBSort( name: str, skel: 'viur.core.skeleton.SkeletonInstance', dbFilter: db.Query, - rawFilter: Dict - ) -> Optional[db.Query]: + rawFilter: dict + ) -> t.Optional[db.Query]: """ Build a DB sort based on the specified name and a raw filter. @@ -286,7 +286,7 @@ def natural_sorting(self, value: str | None) -> str | None: "ẞ": "SS", })) - def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Returns a set of lowercased words that represent searchable tags for the given bone. @@ -305,7 +305,7 @@ def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) result.add(word.lower()) return result - def getUniquePropertyIndexValues(self, skel, name: str) -> List[str]: + def getUniquePropertyIndexValues(self, skel, name: str) -> list[str]: """ Returns a list of unique index values for a given property name. diff --git a/src/viur/core/bones/text.py b/src/viur/core/bones/text.py index 3e7d0989f..ea0780065 100644 --- a/src/viur/core/bones/text.py +++ b/src/viur/core/bones/text.py @@ -8,7 +8,7 @@ from datetime import datetime from html import entities as htmlentitydefs from html.parser import HTMLParser -from typing import Dict, List, Optional, Set, Tuple, Union +import typing as t from viur.core import db, utils from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity @@ -49,7 +49,7 @@ """ -def parseDownloadUrl(urlStr: str) -> Tuple[Optional[str], Optional[bool], Optional[str]]: +def parseDownloadUrl(urlStr: str) -> tuple[t.Optional[str], t.Optional[bool], t.Optional[str]]: """ Parses a file download URL in the format `/file/download/xxxx?sig=yyyy` into its components: blobKey, derived, and filename. If the URL cannot be parsed, the function returns None for each component. @@ -351,9 +351,9 @@ class __undefinedC__: def __init__( self, *, - validHtml: Union[None, Dict] = __undefinedC__, + validHtml: None | dict = __undefinedC__, max_length: int = 200000, - srcSet: Optional[Dict[str, List]] = None, + srcSet: t.Optional[dict[str, list]] = None, indexed: bool = False, **kwargs ): @@ -422,7 +422,7 @@ def isInvalid(self, value): if len(value) > self.max_length: return "Maximum length exceeded" - def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getReferencedBlobs(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Extracts and returns the blob keys of referenced files in the HTML content of the TextBone instance. @@ -479,7 +479,7 @@ def refresh(self, skel, boneName) -> None: elif not self.languages and isinstance(val, str): skel[boneName] = self.singleValueFromClient(val, skel, boneName, None)[0] - def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> Set[str]: + def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) -> set[str]: """ Extracts search tags from the text content of a TextBone. @@ -500,7 +500,7 @@ def getSearchTags(self, skel: 'viur.core.skeleton.SkeletonInstance', name: str) result.add(word.lower()) return result - def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> List[str]: + def getUniquePropertyIndexValues(self, valuesCache: dict, name: str) -> list[str]: """ Retrieves the unique property index values for the TextBone. diff --git a/src/viur/core/cache.py b/src/viur/core/cache.py index d74d09e1e..01305bfac 100644 --- a/src/viur/core/cache.py +++ b/src/viur/core/cache.py @@ -3,7 +3,7 @@ from datetime import timedelta from functools import wraps from hashlib import sha512 -from typing import Callable, Dict, List, Tuple, Union +import typing as t from viur.core import Method, current, db, tasks, utils from viur.core.config import conf @@ -29,8 +29,8 @@ viurCacheName = "viur-cache" -def keyFromArgs(f: Callable, userSensitive: int, languageSensitive: bool, evaluatedArgs: List[str], path: str, - args: Tuple, kwargs: Dict) -> str: +def keyFromArgs(f: t.Callable, userSensitive: int, languageSensitive: bool, evaluatedArgs: list[str], path: str, + args: tuple, kwargs: dict) -> str: """ Utility function to derive a unique but stable string-key that can be used in a datastore-key for the given wrapped function f, the parameter *args and **kwargs it has been called with, @@ -110,8 +110,8 @@ def keyFromArgs(f: Callable, userSensitive: int, languageSensitive: bool, evalua return mysha512.hexdigest() -def wrapCallable(f, urls: List[str], userSensitive: int, languageSensitive: bool, - evaluatedArgs: List[str], maxCacheTime: int): +def wrapCallable(f, urls: list[str], userSensitive: int, languageSensitive: bool, + evaluatedArgs: list[str], maxCacheTime: int): """ Does the actual work of wrapping a callable. Use the decorator enableCache instead of calling this directly. @@ -123,7 +123,7 @@ def wrapCallable(f, urls: List[str], userSensitive: int, languageSensitive: bool f = f._func @wraps(f) - def wrapF(self, *args, **kwargs) -> Union[str, bytes]: + def wrapF(self, *args, **kwargs) -> str | bytes: currReq = current.request.get() if conf.debug.disable_cache or currReq.disableCache: # Caching disabled @@ -174,8 +174,8 @@ def wrapF(self, *args, **kwargs) -> Union[str, bytes]: return method -def enableCache(urls: List[str], userSensitive: int = 0, languageSensitive: bool = False, - evaluatedArgs: Union[List[str], None] = None, maxCacheTime: Union[int, None] = None): +def enableCache(urls: list[str], userSensitive: int = 0, languageSensitive: bool = False, + evaluatedArgs: list[str] | None = None, maxCacheTime: int | None = None): """ Decorator to wrap this cache around a function. In order for this to function correctly, you must provide additional information so ViUR can determine in which situations it's possible to re-use an already cached @@ -211,7 +211,7 @@ def enableCache(urls: List[str], userSensitive: int = 0, languageSensitive: bool @tasks.CallDeferred -def flushCache(prefix: str = None, key: Union[db.Key, None] = None, kind: Union[str, None] = None): +def flushCache(prefix: str = None, key: db.Key | None = None, kind: str | None = None): """ Flushes the cache. Its possible the flush only a part of the cache by specifying the path-prefix. The path is equal to the url that caused it to be cached (eg /page/view) and must be one diff --git a/src/viur/core/config.py b/src/viur/core/config.py index 663ed83e3..fd0245f9e 100644 --- a/src/viur/core/config.py +++ b/src/viur/core/config.py @@ -4,13 +4,13 @@ import os import warnings from pathlib import Path -from typing import Any, Callable, Iterator, Literal, Optional, TYPE_CHECKING, Type, TypeAlias, TypeVar, Union +import typing as t import google.auth from viur.core.version import __version__ -if TYPE_CHECKING: # pragma: no cover +if t.TYPE_CHECKING: # pragma: no cover from viur.core.email import EmailTransport from viur.core.skeleton import SkeletonInstance from viur.core.module import Module @@ -18,8 +18,8 @@ # Construct an alias with a generic type to be able to write Multiple[str] # TODO: Backward compatible implementation, refactor when viur-core # becomes >= Python 3.12 with a type statement (PEP 695) -_T = TypeVar("_T") -Multiple: TypeAlias = list[_T] | tuple[_T] | set[_T] | frozenset[_T] # TODO: Refactor for Python 3.12 +_T = t.TypeVar("_T") +Multiple: t.TypeAlias = list[_T] | tuple[_T] | set[_T] | frozenset[_T] # TODO: Refactor for Python 3.12 class ConfigType: @@ -40,7 +40,7 @@ class ConfigType: def __init__(self, *, strict_mode: bool = None, - parent: Union["ConfigType", None] = None): + parent: t.Union["ConfigType", None] = None): super().__init__() self._strict_mode = strict_mode self._parent = parent @@ -99,7 +99,7 @@ def _resolve_mapping(self, key: str) -> str: def items(self, full_path: bool = False, recursive: bool = True, - ) -> Iterator[tuple[str, Any]]: + ) -> t.Iterator[tuple[str, t.Any]]: """Get all setting of this config as key-value mapping. :param full_path: Show prefix oder only the key. @@ -119,7 +119,7 @@ def items(self, else: yield key, value - def get(self, key: str, default: Any = None) -> Any: + def get(self, key: str, default: t.Any = None) -> t.Any: """Return an item from the config, if it doesn't exist `default` is returned. :param key: The key for the attribute lookup. @@ -136,7 +136,7 @@ def get(self, key: str, default: Any = None) -> Any: except (KeyError, AttributeError): return default - def __getitem__(self, key: str) -> Any: + def __getitem__(self, key: str) -> t.Any: """Support the old dict-like syntax (getter). Not allowed in strict mode. @@ -156,7 +156,7 @@ def __getitem__(self, key: str) -> Any: return getattr(self, key) - def __getattr__(self, key: str) -> Any: + def __getattr__(self, key: str) -> t.Any: """Resolve dot-notation and name mapping in not strict mode. This method is mostly executed by __getitem__, by the @@ -178,7 +178,7 @@ def __getattr__(self, key: str) -> Any: return super().__getattribute__(key) - def __setitem__(self, key: str, value: Any) -> None: + def __setitem__(self, key: str, value: t.Any) -> None: """Support the old dict-like syntax (setter). Not allowed in strict mode. @@ -214,7 +214,7 @@ def __setitem__(self, key: str, value: Any) -> None: return setattr(self, key, value) - def __setattr__(self, key: str, value: Any) -> None: + def __setattr__(self, key: str, value: t.Any) -> None: """Set attributes after applying the old -> new mapping In strict mode it does nothing except a super call @@ -269,7 +269,7 @@ class Admin(ConfigType): color_secondary: str = "#333333" """secondary color for viur-admin""" - module_groups: dict[str, dict[Literal["name", "icon", "sortindex"], str | int]] = {} + module_groups: dict[str, dict[t.Literal["name", "icon", "sortindex"], str | int]] = {} """Module Groups for the VI Group modules in the sidebar in categories (groups). @@ -311,7 +311,7 @@ class Security(ConfigType): """List of URLs for which force_ssl is ignored. Add an asterisk to mark that entry as a prefix (exact match otherwise)""" - content_security_policy: Optional[dict[str, dict[str, list[str]]]] = { + content_security_policy: t.Optional[dict[str, dict[str, list[str]]]] = { "enforce": { "style-src": ["self", "https://accounts.google.com/gsi/style"], "default-src": ["self"], @@ -353,7 +353,7 @@ class Security(ConfigType): enable_coep: bool = False """Shall we emit Cross-Origin-Embedder-Policy: require-corp?""" - enable_coop: Literal[ + enable_coop: t.Literal[ "unsafe-none", "same-origin-allow-popups", "same-origin", "same-origin-plus-COEP"] = "same-origin" """Emit a Cross-Origin-Opener-Policy Header? @@ -361,33 +361,35 @@ class Security(ConfigType): See https://html.spec.whatwg.org/multipage/browsers.html#cross-origin-opener-policy-value """ - enable_corp: Literal["same-origin", "same-site", "cross-origin"] = "same-origin" + enable_corp: t.Literal["same-origin", "same-site", "cross-origin"] = "same-origin" """Emit a Cross-Origin-Resource-Policy Header? See https://fetch.spec.whatwg.org/#cross-origin-resource-policy-header """ - strict_transport_security: Optional[str] = "max-age=22118400" + strict_transport_security: t.Optional[str] = "max-age=22118400" """If set, ViUR will emit a HSTS HTTP-header with each request. Use security.enableStrictTransportSecurity to set this property""" - x_frame_options: Optional[tuple[Literal["deny", "sameorigin", "allow-from"], Optional[str]]] = ("sameorigin", None) + x_frame_options: t.Optional[ + tuple[t.Literal["deny", "sameorigin", "allow-from"], + t.Optional[str]]] = ("sameorigin", None) """If set, ViUR will emit an X-Frame-Options header In case of allow-from, the second parameters must be the host-url. Otherwise, it can be None. """ - x_xss_protection: Optional[bool] = True + x_xss_protection: t.Optional[bool] = True """ViUR will emit an X-XSS-Protection header if set (the default)""" x_content_type_options: bool = True """ViUR will emit X-Content-Type-Options: nosniff Header unless set to False""" - x_permitted_cross_domain_policies: Optional[Literal["none", "master-only", "by-content-type", "all"]] = "none" + x_permitted_cross_domain_policies: t.Optional[t.Literal["none", "master-only", "by-content-type", "all"]] = "none" """Unless set to logical none; ViUR will emit a X-Permitted-Cross-Domain-Policies with each request""" - captcha_default_credentials: Optional[dict[Literal["sitekey", "secret"], str]] = None + captcha_default_credentials: t.Optional[dict[t.Literal["sitekey", "secret"], str]] = None """The default sitekey and secret to use for the captcha-bone. If set, must be a dictionary of "sitekey" and "secret". """ @@ -452,12 +454,12 @@ class Email(ConfigType): log_retention: datetime.timedelta = datetime.timedelta(days=30) """For how long we'll keep successfully send emails in the viur-emails table""" - transport_class: Type["EmailTransport"] = None + transport_class: t.Type["EmailTransport"] = None """Class that actually delivers the email using the service provider of choice. See email.py for more details """ - sendinblue_api_key: Optional[str] = None + sendinblue_api_key: t.Optional[str] = None """API Key for SendInBlue (now Brevo) for the EmailTransportSendInBlue """ @@ -472,7 +474,7 @@ class Email(ConfigType): Otherwise, they'll just be logged. """ - recipient_override: str | list[str] | Callable[[], str | list[str]] | Literal[False] = None + recipient_override: str | list[str] | t.Callable[[], str | list[str]] | t.Literal[False] = None """If set, all outgoing emails will be sent to this address (overriding the 'dests'-parameter in email.sendEmail) """ @@ -480,7 +482,7 @@ class Email(ConfigType): sender_override: str | None = None """If set, this sender will be used, regardless of what the templates advertise as sender""" - admin_recipients: str | list[str] | Callable[[], str | list[str]] = None + admin_recipients: str | list[str] | t.Callable[[], str | list[str]] = None """Sets recipients for mails send with email.sendEMailToAdmins. If not set, all root users will be used.""" _mapping = { @@ -510,7 +512,7 @@ class I18N(ConfigType): language_alias_map: dict[str, str] = {} """Allows mapping of certain languages to one translation (i.e. us->en)""" - language_method: Literal["session", "url", "domain"] = "session" + language_method: t.Literal["session", "url", "domain"] = "session" """Defines how translations are applied: - session: Per Session - url: inject language prefix in url @@ -548,10 +550,10 @@ class User(ConfigType): max_password_length: int = 512 """Prevent Denial of Service attacks using large inputs for pbkdf2""" - otp_issuer: Optional[str] = None + otp_issuer: t.Optional[str] = None """The name of the issuer for the opt token""" - google_client_id: Optional[str] = None + google_client_id: t.Optional[str] = None """OAuth Client ID for Google Login""" google_gsuite_domains: list[str] = [] @@ -591,7 +593,7 @@ class Conf(ConfigType): bone_boolean_str2true: Multiple[str | int] = ("true", "yes", "1") """Allowed values that define a str to evaluate to true""" - cache_environment_key: Optional[Callable[[], str]] = None + cache_environment_key: t.Optional[t.Callable[[], str]] = None """If set, this function will be called for each cache-attempt and the result will be included in the computed cache-key""" @@ -605,7 +607,7 @@ class Conf(ConfigType): db_engine: str = "viur.datastore" """Database engine module""" - error_handler: Callable[[Exception], str] | None = None + error_handler: t.Callable[[Exception], str] | None = None """If set, ViUR calls this function instead of rendering the viur.errorTemplate if an exception occurs""" error_logo: str = None @@ -618,10 +620,10 @@ class Conf(ConfigType): """Hmac-Key used to sign download urls - set automatically""" # TODO: separate this type hints and use it in the File module as well - file_derivations: dict[str, Callable[["SkeletonInstance", dict, dict], list[tuple[str, float, str, Any]]]] = {} + file_derivations: dict[str, t.Callable[["SkeletonInstance", dict, dict], list[tuple[str, float, str, t.Any]]]] = {} """Call-Map for file pre-processors""" - file_thumbnailer_url: Optional[str] = None + file_thumbnailer_url: t.Optional[str] = None # TODO: """docstring""" main_app: "Module" = None @@ -633,25 +635,25 @@ class Conf(ConfigType): max_post_params_count: int = 250 """Upper limit of the amount of parameters we accept per request. Prevents Hash-Collision-Attacks""" - moduleconf_admin_info: dict[str, Any] = { + moduleconf_admin_info: dict[str, t.Any] = { "icon": "icon-settings", "display": "hidden", } """Describing the internal ModuleConfig-module""" - script_admin_info: dict[str, Any] = { + script_admin_info: dict[str, t.Any] = { "icon": "icon-hashtag", "display": "hidden", } """Describing the Script module""" - render_html_download_url_expiration: Optional[float | int] = None + render_html_download_url_expiration: t.Optional[float | int] = None """The default duration, for which downloadURLs generated by the html renderer will stay valid""" - render_json_download_url_expiration: Optional[float | int] = None + render_json_download_url_expiration: t.Optional[float | int] = None """The default duration, for which downloadURLs generated by the json renderer will stay valid""" - request_preprocessor: Optional[Callable[[str], str]] = None + request_preprocessor: t.Optional[t.Callable[[str], str]] = None """Allows the application to register a function that's called before the request gets routed""" search_valid_chars: str = "abcdefghijklmnopqrstuvwxyzäöüß0123456789" @@ -664,7 +666,7 @@ class Conf(ConfigType): ] """Priority, in which skeletons are loaded""" - tasks_custom_environment_handler: tuple[Callable[[], Any], Callable[[Any], None]] = None + tasks_custom_environment_handler: tuple[t.Callable[[], t.Any], t.Callable[[t.Any], None]] = None """ Preserve additional environment in deferred tasks. @@ -682,7 +684,7 @@ class Conf(ConfigType): version: tuple[int, int, int] = tuple(int(part) if part.isdigit() else part for part in __version__.split(".", 3)) """Semantic version number of viur-core as a tuple of 3 (major, minor, patch-level)""" - viur2import_blobsource: Optional[dict[Literal["infoURL", "gsdir"], str]] = None + viur2import_blobsource: t.Optional[dict[t.Literal["infoURL", "gsdir"], str]] = None """Configuration to import file blobs from ViUR2""" def __init__(self, strict_mode: bool = False): diff --git a/src/viur/core/current.py b/src/viur/core/current.py index 1ef5388b7..ebf4e61b2 100644 --- a/src/viur/core/current.py +++ b/src/viur/core/current.py @@ -1,13 +1,13 @@ from contextvars import ContextVar -from typing import Optional, TYPE_CHECKING +import typing as t -if TYPE_CHECKING: +if t.TYPE_CHECKING: from .request import Router from .session import Session from .skeleton import SkeletonInstance -request: ContextVar[Optional["Router"]] = ContextVar("Request", default=None) -request_data: ContextVar[Optional[dict]] = ContextVar("Request-Data", default=None) -session: ContextVar[Optional["Session"]] = ContextVar("Session", default=None) -language: ContextVar[Optional[str]] = ContextVar("Language", default=None) -user: ContextVar[Optional["SkeletonInstance"]] = ContextVar("User", default=None) +request: ContextVar[t.Optional["Router"]] = ContextVar("Request", default=None) +request_data: ContextVar[t.Optional[dict]] = ContextVar("Request-Data", default=None) +session: ContextVar[t.Optional["Session"]] = ContextVar("Session", default=None) +language: ContextVar[t.Optional[str]] = ContextVar("Language", default=None) +user: ContextVar[t.Optional["SkeletonInstance"]] = ContextVar("User", default=None) diff --git a/src/viur/core/decorators.py b/src/viur/core/decorators.py index 023f19351..9e70f0a3b 100644 --- a/src/viur/core/decorators.py +++ b/src/viur/core/decorators.py @@ -1,4 +1,4 @@ -from typing import Callable +import typing as t from viur.core.module import Method __all__ = [ @@ -11,7 +11,7 @@ ] -def exposed(func: Callable) -> Method: +def exposed(func: t.Callable) -> Method: """ Decorator, which marks a function as exposed. @@ -23,7 +23,7 @@ def exposed(func: Callable) -> Method: seo_language_map = func # We received said dictionary: - def expose_with_translations(func: Callable) -> Method: + def expose_with_translations(func: t.Callable) -> Method: func = Method.ensure(func) func.exposed = True func.seo_language_map = seo_language_map @@ -36,7 +36,7 @@ def expose_with_translations(func: Callable) -> Method: return func -def internal_exposed(func: Callable) -> Method: +def internal_exposed(func: t.Callable) -> Method: """ Decorator, which marks a function as internal exposed. """ @@ -45,7 +45,7 @@ def internal_exposed(func: Callable) -> Method: return func -def force_ssl(func: Callable) -> Method: +def force_ssl(func: t.Callable) -> Method: """ Decorator, which enforces usage of an encrypted channel for a given resource. Has no effect on development-servers. @@ -55,7 +55,7 @@ def force_ssl(func: Callable) -> Method: return func -def force_post(func: Callable) -> Method: +def force_post(func: t.Callable) -> Method: """ Decorator, which enforces usage of a http post request. """ @@ -65,10 +65,10 @@ def force_post(func: Callable) -> Method: def access( - *access: str | list[str] | tuple[str] | set[str] | Callable, + *access: str | list[str] | tuple[str] | set[str] | t.Callable, offer_login: bool | str = False, message: str | None = None, -) -> Callable: +) -> t.Callable: """ Decorator, which performs an authentication and authorization check primarily based on the current user's access, which is defined via the `UserSkel.access`-bone. Additionally, a callable for individual access checking can be @@ -106,13 +106,13 @@ def decorator(func): def skey( - func: Callable = None, + func: t.Callable = None, *, - allow_empty: bool | list[str] | tuple[str] | Callable = False, + allow_empty: bool | list[str] | tuple[str] | t.Callable = False, forward_payload: str | None = None, message: str = None, name: str = "skey", - validate: Callable | None = None, + validate: t.Callable | None = None, **extra_kwargs: dict, ) -> Method: """ diff --git a/src/viur/core/email.py b/src/viur/core/email.py index 3c11d3f78..4214e33f1 100644 --- a/src/viur/core/email.py +++ b/src/viur/core/email.py @@ -3,11 +3,13 @@ import logging import os from abc import ABC, abstractmethod -from typing import Any, Callable, Dict, List, Union +import typing as t from urllib import request import requests +if t.TYPE_CHECKING: + from viur.core.skeleton import SkeletonInstance from viur.core import db, utils from viur.core.config import conf from viur.core.tasks import CallDeferred, DeleteEntitiesIter, PeriodicTask @@ -44,9 +46,9 @@ class EmailTransport(ABC): @staticmethod @abstractmethod - def deliverEmail(*, sender: str, dests: List[str], cc: List[str], bcc: List[str], subject: str, body: str, - headers: Dict[str, str], attachments: List[Dict[str, bytes]], - customData: Union[dict, None], **kwargs): + def deliverEmail(*, sender: str, dests: list[str], cc: list[str], bcc: list[str], subject: str, body: str, + headers: dict[str, str], attachments: list[dict[str, bytes]], + customData: dict | None, **kwargs): """ The actual email delivery must be implemented here. All email-adresses can be either in the form of "mm@example.com" or "Max Musterman ". If the delivery was successful, this method @@ -124,7 +126,7 @@ def sendEmailDeferred(emailKey: db.Key): logging.exception(e) -def normalize_to_list(value: Union[None, Any, List[Any], Callable[[], List]]) -> List[Any]: +def normalize_to_list(value: None | t.Any | list[t.Any] | t.Callable[[], list]) -> list[t.Any]: """ Convert the given value to a list. @@ -142,14 +144,14 @@ def normalize_to_list(value: Union[None, Any, List[Any], Callable[[], List]]) -> def sendEMail(*, tpl: str = None, stringTemplate: str = None, - skel: Union[None, Dict, "SkeletonInstance", List["SkeletonInstance"]] = None, + skel: t.Union[None, dict, "SkeletonInstance", list["SkeletonInstance"]] = None, sender: str = None, - dests: Union[str, List[str]] = None, - cc: Union[str, List[str]] = None, - bcc: Union[str, List[str]] = None, - headers: Dict[str, str] = None, - attachments: List[Dict[str, Any]] = None, - context: Union[db.DATASTORE_BASE_TYPES, List[db.DATASTORE_BASE_TYPES], db.Entity] = None, + dests: str | list[str] = None, + cc: str | list[str] = None, + bcc: str | list[str] = None, + headers: dict[str, str] = None, + attachments: list[dict[str, t.Any]] = None, + context: db.DATASTORE_BASE_TYPES | list[db.DATASTORE_BASE_TYPES] | db.Entity = None, **kwargs) -> bool: """ General purpose function for sending e-mail. @@ -302,7 +304,7 @@ class EmailTransportSendInBlue(EmailTransport): "xls", "xlsx", "ppt", "tar", "ez"} @staticmethod - def splitAddress(address: str) -> Dict[str, str]: + def splitAddress(address: str) -> dict[str, str]: """ Splits an Name/Address Pair as "Max Musterman " into a dict {"name": "Max Mustermann", "email": "mm@example.com"} @@ -319,8 +321,8 @@ def splitAddress(address: str) -> Dict[str, str]: return {"email": address} @staticmethod - def deliverEmail(*, sender: str, dests: List[str], cc: List[str], bcc: List[str], subject: str, body: str, - headers: Dict[str, str], attachments: List[Dict[str, bytes]], **kwargs): + def deliverEmail(*, sender: str, dests: list[str], cc: list[str], bcc: list[str], subject: str, body: str, + headers: dict[str, str], attachments: list[dict[str, bytes]], **kwargs): """ Internal function for delivering Emails using Send in Blue. This function requires the conf.email.sendinblue_api_key to be set. diff --git a/src/viur/core/i18n.py b/src/viur/core/i18n.py index 19aa82448..1377694f7 100644 --- a/src/viur/core/i18n.py +++ b/src/viur/core/i18n.py @@ -1,6 +1,6 @@ import datetime import jinja2.ext as jinja2 -from typing import List, Tuple, Union +import typing as t from viur.core.config import conf from viur.core import db, utils, languages, current @@ -15,7 +15,7 @@ class LanguageWrapper(dict): guess the correct language. """ - def __init__(self, languages: Union[List[str], Tuple[str]]): + def __init__(self, languages: list[str] | tuple[str]): super(LanguageWrapper, self).__init__() self.languages = languages @@ -27,7 +27,7 @@ def __bool__(self) -> bool: # (otherwise that test is always true as this dict contains keys) return bool(str(self)) - def resolve(self) -> Union[str, List[str]]: + def resolve(self) -> str | list[str]: """ Causes this wrapper to evaluate to the best language available for the current request. diff --git a/src/viur/core/module.py b/src/viur/core/module.py index 904e13a43..8891c6a09 100644 --- a/src/viur/core/module.py +++ b/src/viur/core/module.py @@ -1,7 +1,7 @@ import copy import inspect import types -import typing +import typing as t import logging from viur.core import db, errors, current, utils from viur.core.config import conf @@ -13,7 +13,7 @@ class Method: """ @classmethod - def ensure(cls, func: typing.Callable | "Method") -> "Method": + def ensure(cls, func: t.Callable | "Method") -> "Method": """ Ensures the provided `func` parameter is either a Method already, or turns it into a Method. This is done to avoid stacking Method objects, which may create @@ -24,7 +24,7 @@ def ensure(cls, func: typing.Callable | "Method") -> "Method": return cls(func) - def __init__(self, func: typing.Callable): + def __init__(self, func: t.Callable): # Content self._func = func self.__name__ = func.__name__ @@ -76,7 +76,7 @@ def __call__(self, *args, **kwargs): if trace := conf.debug.trace: logging.debug(f"calling {self._func=} with raw {args=}, {kwargs=}") - def parse_value_by_annotation(annotation: type, name: str, value: str | list | tuple) -> typing.Any: + def parse_value_by_annotation(annotation: type, name: str, value: str | list | tuple) -> t.Any: """ Tries to parse a value according to a given type. May be called recursively to handle unions, lists and tuples as well. @@ -94,7 +94,7 @@ def parse_value_by_annotation(annotation: type, name: str, value: str | list | t return None # complex types - origin_type = typing.get_origin(annotation) + origin_type = t.get_origin(annotation) if origin_type is list and len(annotation.__args__) == 1: if not isinstance(value, list): @@ -108,13 +108,13 @@ def parse_value_by_annotation(annotation: type, name: str, value: str | list | t return tuple(parse_value_by_annotation(annotation.__args__[0], name, item) for item in value) - elif origin_type is typing.Literal: + elif origin_type is t.Literal: if not any(value == str(literal) for literal in annotation.__args__): raise errors.NotAcceptable(f"Expecting any of {annotation.__args__} for {name}") return value - elif origin_type is typing.Union or isinstance(annotation, types.UnionType): + elif origin_type is t.Union or isinstance(annotation, types.UnionType): for i, sub_annotation in enumerate(annotation.__args__): try: return parse_value_by_annotation(sub_annotation, name, value) @@ -304,7 +304,7 @@ def describe(self) -> dict: """ Describes the Method with a """ - return_doc = typing.get_type_hints(self._func).get("return") + return_doc = t.get_type_hints(self._func).get("return") ret = { "args": { @@ -349,7 +349,7 @@ class Module: This is the root module prototype that serves a minimal module in the ViUR system without any other bindings. """ - handler: str | typing.Callable = None + handler: str | t.Callable = None """ This is the module's handler, respectively its type. Use the @property-decorator in specific Modules to construct the handler's value dynamically. @@ -410,7 +410,7 @@ class MyOrders(List): Great, this part is now user and robot friendly :) """ - adminInfo: dict[str, typing.Any] | typing.Callable = None + adminInfo: dict[str, t.Any] | t.Callable = None """ This is a ``dict`` holding the information necessary for the Vi/Admin to handle this module. @@ -446,7 +446,7 @@ class MyOrders(List): replaced as needed. If more than one preview-url is needed, supply a dictionary where the key is the URL and the value the description shown to the user. - views: ``List[Dict[str, typing.Any]]`` + views: ``List[Dict[str, t.Any]]`` (Optional) List of nested adminInfo like dictionaries. Used to define additional views on the module. Useful f.e. for an order module, where you want separate list of "payed orders", "unpayed orders", "orders waiting for shipment", etc. If such views are defined, @@ -487,7 +487,7 @@ class MyOrders(List): moduleGroup: ``str`` (Optional) If set, should be a key of a moduleGroup defined in .... . - editViews: ``Dict[str, typing.Any]`` + editViews: ``Dict[str, t.Any]`` (Optional) If set, will embed another list-widget in the edit forms for a given entity. See .... for more details. diff --git a/src/viur/core/modules/file.py b/src/viur/core/modules/file.py index 00205b941..8248187b2 100644 --- a/src/viur/core/modules/file.py +++ b/src/viur/core/modules/file.py @@ -7,7 +7,7 @@ from datetime import datetime, timedelta from io import BytesIO from quopri import decodestring -from typing import Any, List, Tuple, Union +import typing as t from urllib.request import urlopen import email.header @@ -416,7 +416,7 @@ class File(Tree): blobCacheTime = 60 * 60 * 24 # Requests to file/download will be served with cache-control: public, max-age=blobCacheTime if set - def write(self, filename: str, content: Any, mimetype: str = "text/plain", width: int = None, + def write(self, filename: str, content: t.Any, mimetype: str = "text/plain", width: int = None, height: int = None) -> db.Key: """ Write a file from any buffer into the file module. @@ -445,7 +445,7 @@ def write(self, filename: str, content: Any, mimetype: str = "text/plain", width return skel.toDB() - def read(self, key: db.Key | int | str | None = None, path: str | None = None) -> Tuple[io.BytesIO, str]: + def read(self, key: db.Key | int | str | None = None, path: str | None = None) -> tuple[io.BytesIO, str]: """ Read a file from the Cloud Storage. @@ -486,8 +486,8 @@ def deleteRecursive(self, parentKey): if skel.fromDB(d.key): skel.delete() - def signUploadURL(self, mimeTypes: Union[List[str], None] = None, maxSize: Union[int, None] = None, - node: Union[str, None] = None): + def signUploadURL(self, mimeTypes: list[str] | None = None, maxSize: int | None = None, + node: str | None = None): """ Internal helper that will create a signed upload-url that can be used to retrieve an uploadURL from getUploadURL for guests / users without having file/add permissions. This URL is valid for an hour and can @@ -519,8 +519,8 @@ def signUploadURL(self, mimeTypes: Union[List[str], None] = None, maxSize: Union def initializeUpload(self, fileName: str, mimeType: str, - node: Union[str, None], - size: Union[int, None] = None) -> Tuple[str, str]: + node: str | None, + size: int | None = None) -> tuple[str, str]: """ Internal helper that registers a new upload. Will create the pending fileSkel entry (needed to remove any started uploads from GCS if that file isn't used) and creates a resumable (and signed) uploadURL for that. diff --git a/src/viur/core/modules/user.py b/src/viur/core/modules/user.py index 39893f6a3..31bed58e2 100644 --- a/src/viur/core/modules/user.py +++ b/src/viur/core/modules/user.py @@ -13,7 +13,7 @@ import pyotp import base64 import dataclasses -import typing +import typing as t from google.auth.transport import requests from google.oauth2 import id_token @@ -697,7 +697,7 @@ class for configuration. """ secret: str timedrift: float = 0.0 - algorithm: typing.Literal["sha1", "sha256"] = "sha1" + algorithm: t.Literal["sha1", "sha256"] = "sha1" interval: int = 60 class OtpSkel(skeleton.RelSkel): diff --git a/src/viur/core/pagination.py b/src/viur/core/pagination.py index 121ab0966..24adcc94c 100644 --- a/src/viur/core/pagination.py +++ b/src/viur/core/pagination.py @@ -1,5 +1,5 @@ from hashlib import sha256 -from typing import List, Optional +import typing as t from viur.core import db, utils @@ -65,7 +65,7 @@ def key_from_query(self, query: db.Query) -> str: filter_key = "".join("%s%s" % (x, y) for x, y in orig_filter) return sha256(filter_key.encode()).hexdigest() - def get_or_build_index(self, orig_query: db.Query) -> List[str]: + def get_or_build_index(self, orig_query: db.Query) -> list[str]: """ Builds a specific index based on origQuery AND local variables (self.page_size and self.max_pages) @@ -102,7 +102,7 @@ def get_or_build_index(self, orig_query: db.Query) -> List[str]: db.Put(entry) return cursors - def cursor_for_query(self, query: db.Query, page: int) -> Optional[str]: + def cursor_for_query(self, query: db.Query, page: int) -> t.Optional[str]: """ Returns the starting-cursor for the given query and page using an index. @@ -123,7 +123,7 @@ def cursor_for_query(self, query: db.Query, page: int) -> Optional[str]: else: return None - def get_pages(self, query: db.Query) -> List[str]: + def get_pages(self, query: db.Query) -> list[str]: """ Returns a list of all starting-cursors for this query. The first element is always None as the first page doesn't diff --git a/src/viur/core/prototypes/list.py b/src/viur/core/prototypes/list.py index 9c99e26af..87a7cc46a 100644 --- a/src/viur/core/prototypes/list.py +++ b/src/viur/core/prototypes/list.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Optional +import typing as t from viur.core import current, db, errors, utils from viur.core.decorators import * from viur.core.cache import flushCache @@ -78,7 +78,7 @@ def editSkel(self, *args, **kwargs) -> SkeletonInstance: @exposed @force_post @skey - def preview(self, *args, **kwargs) -> Any: + def preview(self, *args, **kwargs) -> t.Any: """ Renders data for an entry, without reading from the database. This function allows to preview an entry without writing it to the database. @@ -98,7 +98,7 @@ def preview(self, *args, **kwargs) -> Any: return self.render.view(skel) @exposed - def structure(self, *args, **kwargs) -> Any: + def structure(self, *args, **kwargs) -> t.Any: """ :returns: Returns the structure of our skeleton as used in list/view. Values are the defaultValues set in each bone. @@ -114,7 +114,7 @@ def structure(self, *args, **kwargs) -> Any: return self.render.view(skel) @exposed - def view(self, key: db.Key | int | str, *args, **kwargs) -> Any: + def view(self, key: db.Key | int | str, *args, **kwargs) -> t.Any: """ Prepares and renders a single entry for viewing. @@ -141,7 +141,7 @@ def view(self, key: db.Key | int | str, *args, **kwargs) -> Any: return self.render.view(skel) @exposed - def list(self, *args, **kwargs) -> Any: + def list(self, *args, **kwargs) -> t.Any: """ Prepares and renders a list of entries. @@ -166,7 +166,7 @@ def list(self, *args, **kwargs) -> Any: @force_ssl @exposed @skey(allow_empty=True) - def edit(self, key: db.Key | int | str, *args, **kwargs) -> Any: + def edit(self, key: db.Key | int | str, *args, **kwargs) -> t.Any: """ Modify an existing entry, and render the entry, eventually with error notes on incorrect data. Data is taken by any other arguments in *kwargs*. @@ -209,7 +209,7 @@ def edit(self, key: db.Key | int | str, *args, **kwargs) -> Any: @force_ssl @exposed @skey(allow_empty=True) - def add(self, *args, **kwargs) -> Any: + def add(self, *args, **kwargs) -> t.Any: """ Add a new entry, and render the entry, eventually with error notes on incorrect data. Data is taken by any other arguments in *kwargs*. @@ -247,7 +247,7 @@ def add(self, *args, **kwargs) -> Any: @force_post @exposed @skey - def delete(self, key: db.Key | int | str, *args, **kwargs) -> Any: + def delete(self, key: db.Key | int | str, *args, **kwargs) -> t.Any: """ Delete an entry. @@ -275,7 +275,7 @@ def delete(self, key: db.Key | int | str, *args, **kwargs) -> Any: return self.render.deleteSuccess(skel) @exposed - def index(self, *args, **kwargs) -> Any: + def index(self, *args, **kwargs) -> t.Any: """ Default, SEO-Friendly fallback for view and list. @@ -308,7 +308,7 @@ def getDefaultListParams(self): ## Default access control functions - def listFilter(self, query: db.Query) -> Optional[db.Query]: + def listFilter(self, query: db.Query) -> t.Optional[db.Query]: """ Access control function on item listing. diff --git a/src/viur/core/prototypes/singleton.py b/src/viur/core/prototypes/singleton.py index 5059b56d4..4028ae52b 100644 --- a/src/viur/core/prototypes/singleton.py +++ b/src/viur/core/prototypes/singleton.py @@ -1,5 +1,5 @@ import logging -from typing import Any +import typing as t from viur.core import db, current, utils, errors from viur.core.decorators import * from viur.core.cache import flushCache @@ -57,7 +57,7 @@ def editSkel(self, *args, **kwargs) -> SkeletonInstance: @exposed @skey - def preview(self, *args, **kwargs) -> Any: + def preview(self, *args, **kwargs) -> t.Any: """ Renders data for the entry, without reading it from the database. This function allows to preview the entry without writing it to the database. @@ -77,7 +77,7 @@ def preview(self, *args, **kwargs) -> Any: return self.render.view(skel) @exposed - def structure(self, *args, **kwargs) -> Any: + def structure(self, *args, **kwargs) -> t.Any: """ :returns: Returns the structure of our skeleton as used in list/view. Values are the defaultValues set in each bone. @@ -90,7 +90,7 @@ def structure(self, *args, **kwargs) -> Any: return self.render.view(skel) @exposed - def view(self, *args, **kwargs) -> Any: + def view(self, *args, **kwargs) -> t.Any: """ Prepares and renders the singleton entry for viewing. @@ -119,7 +119,7 @@ def view(self, *args, **kwargs) -> Any: @exposed @force_ssl @skey(allow_empty=True) - def edit(self, *args, **kwargs) -> Any: + def edit(self, *args, **kwargs) -> t.Any: """ Modify the existing entry, and render the entry, eventually with error notes on incorrect data. diff --git a/src/viur/core/prototypes/skelmodule.py b/src/viur/core/prototypes/skelmodule.py index d9893d918..6a92c7780 100644 --- a/src/viur/core/prototypes/skelmodule.py +++ b/src/viur/core/prototypes/skelmodule.py @@ -1,6 +1,6 @@ from viur.core import Module from viur.core.skeleton import skeletonByKind, Skeleton, SkeletonInstance -from typing import Tuple, Type +import typing as t class SkelModule(Module): @@ -20,7 +20,7 @@ class SkelModule(Module): For more information, refer to the function :func:`~_resolveSkelCls`. """ - def _resolveSkelCls(self, *args, **kwargs) -> Type[Skeleton]: + def _resolveSkelCls(self, *args, **kwargs) -> t.Type[Skeleton]: """ Retrieve the generally associated :class:`viur.core.skeleton.Skeleton` that is used by the application. diff --git a/src/viur/core/prototypes/tree.py b/src/viur/core/prototypes/tree.py index c876dc235..eb5e5e0b9 100644 --- a/src/viur/core/prototypes/tree.py +++ b/src/viur/core/prototypes/tree.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Dict, List, Literal, Optional, Type +import typing as t from viur.core import utils, errors, db, current from viur.core.decorators import * from viur.core.bones import KeyBone, SortIndexBone @@ -9,7 +9,7 @@ from .skelmodule import SkelModule -SkelType = Literal["node", "leaf"] +SkelType = t.Literal["node", "leaf"] class TreeSkel(Skeleton): @@ -55,7 +55,7 @@ def __init__(self, moduleName, modulePath, *args, **kwargs): def handler(self): return "tree" if self.leafSkelCls else "tree.node" # either a tree or a tree with nodes only (former hierarchy) - def _checkSkelType(self, skelType: Any) -> Optional[SkelType]: + def _checkSkelType(self, skelType: t.Any) -> t.Optional[SkelType]: """ Checks for correct skelType. @@ -67,7 +67,7 @@ def _checkSkelType(self, skelType: Any) -> Optional[SkelType]: return None - def _resolveSkelCls(self, skelType: SkelType, *args, **kwargs) -> Type[Skeleton]: + def _resolveSkelCls(self, skelType: SkelType, *args, **kwargs) -> t.Type[Skeleton]: if not (skelType := self._checkSkelType(skelType)): raise ValueError("Unsupported skelType") @@ -134,7 +134,7 @@ def ensureOwnModuleRootNode(self) -> db.Entity: kindName = self.viewSkel("node").kindName return db.GetOrInsert(db.Key(kindName, key), creationdate=utils.utcNow(), rootNode=1) - def getAvailableRootNodes(self, *args, **kwargs) -> List[Dict[Literal["name", "key"], str]]: + def getAvailableRootNodes(self, *args, **kwargs) -> list[dict[t.Literal["name", "key"], str]]: """ Default function for providing a list of root node items. This list is requested by several module-internal functions and *must* be @@ -241,7 +241,7 @@ def pathToKey(self, key: db.Key): ## External exposed functions @exposed - def listRootNodes(self, *args, **kwargs) -> Any: + def listRootNodes(self, *args, **kwargs) -> t.Any: """ Renders a list of all available repositories for the current user using the modules default renderer. @@ -251,7 +251,7 @@ def listRootNodes(self, *args, **kwargs) -> Any: return self.render.listRootNodes(self.getAvailableRootNodes(*args, **kwargs)) @exposed - def list(self, skelType: SkelType, *args, **kwargs) -> Any: + def list(self, skelType: SkelType, *args, **kwargs) -> t.Any: """ Prepares and renders a list of entries. @@ -277,7 +277,7 @@ def list(self, skelType: SkelType, *args, **kwargs) -> Any: return self.render.list(res) @exposed - def structure(self, skelType: SkelType, *args, **kwargs) -> Any: + def structure(self, skelType: SkelType, *args, **kwargs) -> t.Any: """ :returns: Returns the structure of our skeleton as used in list/view. Values are the defaultValues set in each bone. @@ -296,7 +296,7 @@ def structure(self, skelType: SkelType, *args, **kwargs) -> Any: return self.render.view(skel) @exposed - def view(self, skelType: SkelType, key: db.Key | int | str, *args, **kwargs) -> Any: + def view(self, skelType: SkelType, key: db.Key | int | str, *args, **kwargs) -> t.Any: """ Prepares and renders a single entry for viewing. @@ -330,7 +330,7 @@ def view(self, skelType: SkelType, key: db.Key | int | str, *args, **kwargs) -> @exposed @force_ssl @skey(allow_empty=True) - def add(self, skelType: SkelType, node: db.Key | int | str, *args, **kwargs) -> Any: + def add(self, skelType: SkelType, node: db.Key | int | str, *args, **kwargs) -> t.Any: """ Add a new entry with the given parent *node*, and render the entry, eventually with error notes on incorrect data. Data is taken by any other arguments in *kwargs*. @@ -380,7 +380,7 @@ def add(self, skelType: SkelType, node: db.Key | int | str, *args, **kwargs) -> @exposed @force_ssl @skey(allow_empty=True) - def edit(self, skelType: SkelType, key: db.Key | int | str, *args, **kwargs) -> Any: + def edit(self, skelType: SkelType, key: db.Key | int | str, *args, **kwargs) -> t.Any: """ Modify an existing entry, and render the entry, eventually with error notes on incorrect data. Data is taken by any other arguments in *kwargs*. @@ -427,7 +427,7 @@ def edit(self, skelType: SkelType, key: db.Key | int | str, *args, **kwargs) -> @force_ssl @force_post @skey - def delete(self, skelType: SkelType, key: str, *args, **kwargs) -> Any: + def delete(self, skelType: SkelType, key: str, *args, **kwargs) -> t.Any: """ Deletes an entry or an directory (including its contents). @@ -570,7 +570,7 @@ def move(self, skelType: SkelType, key: db.Key | int | str, parentNode: str, *ar ## Default access control functions - def listFilter(self, query: db.Query) -> Optional[db.Query]: + def listFilter(self, query: db.Query) -> t.Optional[db.Query]: """ Access control function on item listing. @@ -603,7 +603,7 @@ def canView(self, skelType: SkelType, skel: SkeletonInstance) -> bool: return False return True - def canAdd(self, skelType: SkelType, parentNodeSkel: Optional[SkeletonInstance]) -> bool: + def canAdd(self, skelType: SkelType, parentNodeSkel: t.Optional[SkeletonInstance]) -> bool: """ Access control function for adding permission. diff --git a/src/viur/core/ratelimit.py b/src/viur/core/ratelimit.py index a8d19f09e..045ae5119 100644 --- a/src/viur/core/ratelimit.py +++ b/src/viur/core/ratelimit.py @@ -1,6 +1,6 @@ from viur.core import current, db, errors, utils from viur.core.tasks import PeriodicTask, DeleteEntitiesIter -from typing import Literal, Union +import typing as t from datetime import timedelta @@ -15,7 +15,7 @@ class RateLimit(object): """ rateLimitKind = "viur-ratelimit" - def __init__(self, resource: str, maxRate: int, minutes: int, method: Literal["ip", "user"]): + def __init__(self, resource: str, maxRate: int, minutes: int, method: t.Literal["ip", "user"]): """ Initializes a new RateLimit gate. @@ -33,7 +33,7 @@ def __init__(self, resource: str, maxRate: int, minutes: int, method: Literal["i assert method in ["ip", "user"], "method must be 'ip' or 'user'" self.useUser = method == "user" - def _getEndpointKey(self) -> Union[db.Key, str]: + def _getEndpointKey(self) -> db.Key | str: """ :warning: It's invalid to call _getEndpointKey if method is set to user and there's no user logged in! diff --git a/src/viur/core/render/html/default.py b/src/viur/core/render/html/default.py index 3caada47f..d74ca3908 100644 --- a/src/viur/core/render/html/default.py +++ b/src/viur/core/render/html/default.py @@ -4,7 +4,7 @@ import functools import logging import os -from typing import Any, Dict, List, Literal, Tuple, Union +import typing as t from jinja2 import ChoiceLoader, Environment, FileSystemLoader, Template @@ -66,7 +66,7 @@ def __init__(self, parent=None, *args, **kwargs): def getTemplateFileName( self, - template: str | List[str] | Tuple[str], + template: str | list[str] | tuple[str], ignoreStyle: bool = False, raise_exception: bool = True, ) -> str | None: @@ -152,10 +152,10 @@ def getLoaders(self) -> ChoiceLoader: def renderBoneValue(self, bone: BaseBone, skel: SkeletonInstance, - key: Any, # TODO: unused - boneValue: Any, + key: t.Any, # TODO: unused + boneValue: t.Any, isLanguageWrapped: bool = False - ) -> Union[List, Dict, KeyValueWrapper, LanguageWrapper, str, None]: + ) -> list | dict | KeyValueWrapper | LanguageWrapper | str | None: """ Renders the value of a bone. @@ -253,7 +253,7 @@ def render_action_template( skel: SkeletonInstance, action: str, tpl: str = None, - params: Dict = None, + params: dict = None, **kwargs ) -> str: """ @@ -301,7 +301,7 @@ def render_view_template( skel: SkeletonInstance, action: str, tpl: str = None, - params: Dict = None, + params: dict = None, **kwargs ) -> str: """ @@ -329,7 +329,7 @@ def render_view_template( **kwargs ) - def list(self, skellist: SkelList, action: str = "list", tpl: str = None, params: Any = None, **kwargs) -> str: + def list(self, skellist: SkelList, action: str = "list", tpl: str = None, params: t.Any = None, **kwargs) -> str: """ Renders a page with a list of entries. @@ -350,7 +350,8 @@ def list(self, skellist: SkelList, action: str = "list", tpl: str = None, params return template.render(skellist=skellist, action=action, params=params, **kwargs) - def view(self, skel: SkeletonInstance, action: str = "view", tpl: str = None, params: Any = None, **kwargs) -> str: + def view(self, skel: SkeletonInstance, action: str = "view", tpl: str = None, params: t.Any = None, + **kwargs) -> str: """ Renders a page for viewing an entry. @@ -358,7 +359,7 @@ def view(self, skel: SkeletonInstance, action: str = "view", tpl: str = None, pa """ return self.render_view_template("view", skel, action, tpl, params, **kwargs) - def add(self, skel: SkeletonInstance, action: str = "add", tpl: str = None, params: Any = None, **kwargs) -> str: + def add(self, skel: SkeletonInstance, action: str = "add", tpl: str = None, params: t.Any = None, **kwargs) -> str: """ Renders a page for adding an entry. @@ -366,7 +367,8 @@ def add(self, skel: SkeletonInstance, action: str = "add", tpl: str = None, para """ return self.render_action_template("add", skel, action, tpl, params, **kwargs) - def edit(self, skel: SkeletonInstance, action: str = "edit", tpl: str = None, params: Any = None, **kwargs) -> str: + def edit(self, skel: SkeletonInstance, action: str = "edit", tpl: str = None, params: t.Any = None, + **kwargs) -> str: """ Renders a page for modifying an entry. @@ -379,7 +381,7 @@ def addSuccess( skel: SkeletonInstance, action: str = "addSuccess", tpl: str = None, - params: Any = None, + params: t.Any = None, **kwargs ) -> str: """ @@ -394,7 +396,7 @@ def editSuccess( skel: SkeletonInstance, action: str = "editSuccess", tpl: str = None, - params: Any = None, + params: t.Any = None, **kwargs ) -> str: """ @@ -409,7 +411,7 @@ def deleteSuccess( skel: SkeletonInstance, action: str = "deleteSuccess", tpl: str = None, - params: Any = None, + params: t.Any = None, **kwargs ) -> str: """ @@ -421,10 +423,10 @@ def deleteSuccess( def listRootNodes( # fixme: This is a relict, should be solved differently (later!). self, - repos: List[Dict[Literal["key", "name"], Any]], + repos: t.List[dict[t.Literal["key", "name"], t.Any]], action: str = "listrootnodes", tpl: str = None, - params: Any = None, + params: t.Any = None, **kwargs ) -> str: """ @@ -443,11 +445,11 @@ def listRootNodes( # fixme: This is a relict, should be solved differently (lat return template.render(repos=repos, action=action, params=params, **kwargs) def renderEmail(self, - dests: List[str], + dests: t.List[str], file: str = None, template: str = None, - skel: Union[None, Dict, SkeletonInstance, List[SkeletonInstance]] = None, - **kwargs) -> Tuple[str, str]: + skel: None | dict | SkeletonInstance | t.List["SkeletonInstance"] = None, + **kwargs) -> tuple[str, str]: """ Renders an email. Uses the first not-empty line as subject and the remaining template as body. diff --git a/src/viur/core/render/html/env/date.py b/src/viur/core/render/html/env/date.py index 708e7e4fb..8333084f2 100644 --- a/src/viur/core/render/html/env/date.py +++ b/src/viur/core/render/html/env/date.py @@ -1,4 +1,4 @@ -from typing import Type +import typing as t from datetime import date as date_orig, datetime as datetime_orig, time as time_orig, timedelta as timedelta_orig @@ -7,7 +7,7 @@ @jinjaGlobalFunction -def dateTime(render: Render) -> Type[datetime_orig]: +def dateTime(render: Render) -> t.Type[datetime_orig]: """ Jinja2 global: Returns the datetime class @@ -17,7 +17,7 @@ def dateTime(render: Render) -> Type[datetime_orig]: @jinjaGlobalFunction -def date(render: Render) -> Type[date_orig]: +def date(render: Render) -> t.Type[date_orig]: """ Jinja2 global: Returns the date class @@ -27,7 +27,7 @@ def date(render: Render) -> Type[date_orig]: @jinjaGlobalFunction -def time(render: Render) -> Type[time_orig]: +def time(render: Render) -> t.Type[time_orig]: """ Jinja2 global: Returns the time class @@ -37,7 +37,7 @@ def time(render: Render) -> Type[time_orig]: @jinjaGlobalFunction -def timedelta(render: Render) -> Type[timedelta_orig]: +def timedelta(render: Render) -> t.Type[timedelta_orig]: """ Jinja2 global: Returns the timedelta class diff --git a/src/viur/core/render/html/env/debug.py b/src/viur/core/render/html/env/debug.py index f55d9f993..7f97a16d6 100644 --- a/src/viur/core/render/html/env/debug.py +++ b/src/viur/core/render/html/env/debug.py @@ -1,7 +1,7 @@ from logging import critical, debug, error, info, warning import pprint -from typing import Any +import typing as t from ..utils import jinjaGlobalFunction from ..default import Render @@ -34,7 +34,7 @@ def logging(render: Render, msg: str, level: str = "info", *args, **kwargs) -> N @jinjaGlobalFunction -def pprint(render: Render, obj: Any) -> str: +def pprint(render: Render, obj: t.Any) -> str: """ Jinja2 global: Provides a pprint function that renders into HTML. The function shall be used for debug purposes. diff --git a/src/viur/core/render/html/env/regex.py b/src/viur/core/render/html/env/regex.py index 53f673798..6953f07bd 100644 --- a/src/viur/core/render/html/env/regex.py +++ b/src/viur/core/render/html/env/regex.py @@ -1,5 +1,5 @@ import re -from typing import Optional +import typing as t from viur.core.render.html.utils import jinjaGlobalFunction from ..default import Render @@ -37,7 +37,7 @@ def regexReplace(render: Render, string: str, pattern: str, replace: str) -> str @jinjaGlobalFunction -def regexSearch(render: Render, string: str, pattern: str, flags=0) -> Optional[re.Match]: +def regexSearch(render: Render, string: str, pattern: str, flags=0) -> t.Optional[re.Match]: """ Jinja2 global: Search a string for regular expression pattern. This function internally runs re.search(). diff --git a/src/viur/core/render/html/env/session.py b/src/viur/core/render/html/env/session.py index 7b267a90f..eb14320ac 100644 --- a/src/viur/core/render/html/env/session.py +++ b/src/viur/core/render/html/env/session.py @@ -1,4 +1,4 @@ -from typing import Dict +import typing as t from viur.core.render.html.utils import jinjaGlobalFunction from viur.core import current @@ -6,7 +6,7 @@ @jinjaGlobalFunction -def getSession(render: Render) -> Dict: +def getSession(render: Render) -> dict: """ Jinja2 global: Allows templates to store variables server-side inside the session. diff --git a/src/viur/core/render/html/env/strings.py b/src/viur/core/render/html/env/strings.py index ec5a08894..ee43fba59 100644 --- a/src/viur/core/render/html/env/strings.py +++ b/src/viur/core/render/html/env/strings.py @@ -1,12 +1,12 @@ import json -from typing import Any +import typing as t from ..utils import jinjaGlobalFilter from ..default import Render @jinjaGlobalFilter -def parseJSON(render: Render, value: str) -> Any: +def parseJSON(render: Render, value: str) -> t.Any: """ Jinja2 filter: Parses a JSON-string into a python object. diff --git a/src/viur/core/render/html/env/tests.py b/src/viur/core/render/html/env/tests.py index f622612f8..dcf43fb14 100644 --- a/src/viur/core/render/html/env/tests.py +++ b/src/viur/core/render/html/env/tests.py @@ -1,16 +1,16 @@ -from typing import Any +import typing as t from ..default import Render from ..utils import jinjaGlobalTest @jinjaGlobalTest("dict") -def test_dict(render: Render, value: Any) -> bool: +def test_dict(render: Render, value: t.Any) -> bool: """Jinja2 test: Return True if the object is a dict.""" return isinstance(value, dict) @jinjaGlobalTest("list") -def test_list(render: Render, value: Any) -> bool: +def test_list(render: Render, value: t.Any) -> bool: """Jinja2 test: Return True if the object is a list.""" return isinstance(value, list) diff --git a/src/viur/core/render/html/env/viur.py b/src/viur/core/render/html/env/viur.py index 78079ea1c..66be83103 100644 --- a/src/viur/core/render/html/env/viur.py +++ b/src/viur/core/render/html/env/viur.py @@ -6,7 +6,7 @@ from collections import OrderedDict from datetime import timedelta from hashlib import sha512 -from typing import Any, Dict, List, NoReturn, Optional, Union +import typing as t from qrcode import make as qrcode_make from qrcode.image import svg as qrcode_svg @@ -29,7 +29,7 @@ def translate(render: Render, key: str, **kwargs) -> str: @jinjaGlobalFunction -def execRequest(render: Render, path: str, *args, **kwargs) -> Any: +def execRequest(render: Render, path: str, *args, **kwargs) -> t.Any: """ Jinja2 global: Perform an internal Request. @@ -115,7 +115,7 @@ def execRequest(render: Render, path: str, *args, **kwargs) -> Any: @jinjaGlobalFunction -def getCurrentUser(render: Render) -> Optional[SkeletonInstance]: +def getCurrentUser(render: Render) -> t.Optional[SkeletonInstance]: """ Jinja2 global: Returns the current user from the session, or None if not logged in. @@ -129,7 +129,7 @@ def getCurrentUser(render: Render) -> Optional[SkeletonInstance]: @jinjaGlobalFunction -def getSkel(render: Render, module: str, key: str = None, skel: str = "viewSkel") -> Union[dict, bool, None]: +def getSkel(render: Render, module: str, key: str = None, skel: str = "viewSkel") -> dict | bool | None: """ Jinja2 global: Fetch an entry from a given module, and return the data as a dict, prepared for direct use in the output. @@ -236,7 +236,7 @@ def getAppVersion(render: Render) -> str: @jinjaGlobalFunction -def redirect(render: Render, url: str) -> NoReturn: +def redirect(render: Render, url: str) -> t.NoReturn: """ Jinja2 global: Redirect to another URL. @@ -287,7 +287,7 @@ def modulePath(render: Render) -> str: @jinjaGlobalFunction def getList(render: Render, module: str, skel: str = "viewSkel", - _noEmptyFilter: bool = False, *args, **kwargs) -> Union[bool, None, List[SkeletonInstance]]: + _noEmptyFilter: bool = False, *args, **kwargs) -> bool | None | list[SkeletonInstance]: """ Jinja2 global: Fetches a list of entries which match the given filter criteria. @@ -335,7 +335,7 @@ def getSecurityKey(render: Render, **kwargs) -> str: def getStructure(render: Render, module: str, skel: str = "viewSkel", - subSkel: Optional[str] = None) -> Union[Dict, bool]: + subSkel: t.Optional[str] = None) -> dict | bool: """ Jinja2 global: Returns the skeleton structure instead of data for a module. @@ -367,7 +367,7 @@ def getStructure(render: Render, @jinjaGlobalFunction -def requestParams(render: Render) -> Dict[str, str]: +def requestParams(render: Render) -> dict[str, str]: """ Jinja2 global: Allows for accessing the request-parameters from the template. @@ -408,7 +408,7 @@ def updateURL(render: Render, **kwargs) -> str: @jinjaGlobalFilter -def fileSize(render: Render, value: Union[int, float], binary: bool = False) -> str: +def fileSize(render: Render, value: int | float, binary: bool = False) -> str: """ Jinja2 filter: Format the value in an 'human-readable' file size (i.e. 13 kB, 4.1 MB, 102 Bytes, etc). Per default, decimal prefixes are used (Mega, Giga, etc.). When the second parameter is set to True, @@ -496,7 +496,7 @@ def className(render: Render, s: str) -> str: @jinjaGlobalFilter -def shortKey(render: Render, val: str) -> Optional[str]: +def shortKey(render: Render, val: str) -> t.Optional[str]: """ Jinja2 filter: Make a shortkey from an entity-key. @@ -551,11 +551,11 @@ def renderEditBone(render: Render, skel, boneName, boneErrors=None, prefix=None) @jinjaGlobalFunction def renderEditForm(render: Render, - skel: Dict, - ignore: List[str] = None, - hide: List[str] = None, + skel: dict, + ignore: list[str] = None, + hide: list[str] = None, prefix=None, - bones: List[str] = None, + bones: list[str] = None, ) -> str: """Render an edit-form based on a skeleton. @@ -636,7 +636,7 @@ def renderEditForm(render: Render, @jinjaGlobalFunction -def embedSvg(render: Render, name: str, classes: Union[List[str], None] = None, **kwargs: Dict[str, str]) -> str: +def embedSvg(render: Render, name: str, classes: list[str] | None = None, **kwargs: dict[str, str]) -> str: """ jinja2 function to get an -tag for a SVG. This method will not check the existence of a SVG! @@ -667,9 +667,9 @@ def embedSvg(render: Render, name: str, classes: Union[List[str], None] = None, @jinjaGlobalFunction def downloadUrlFor(render: Render, fileObj: dict, - expires: Union[None, int] = conf.render_html_download_url_expiration, - derived: Optional[str] = None, - downloadFileName: Optional[str] = None) -> Optional[str]: + expires: None | int = conf.render_html_download_url_expiration, + derived: t.Optional[str] = None, + downloadFileName: t.Optional[str] = None) -> t.Optional[str]: """ Constructs a signed download-url for the given file-bone. Mostly a wrapper around :meth:`viur.core.utils.downloadUrlFor`. @@ -702,8 +702,8 @@ def downloadUrlFor(render: Render, @jinjaGlobalFunction -def srcSetFor(render: Render, fileObj: dict, expires: Optional[int], - width: Optional[int] = None, height: Optional[int] = None) -> str: +def srcSetFor(render: Render, fileObj: dict, expires: t.Optional[int], + width: t.Optional[int] = None, height: t.Optional[int] = None) -> str: """ Generates a string suitable for use as the srcset tag in html. This functionality provides the browser with a list of images in different sizes and allows it to choose the smallest file that will fill it's viewport without diff --git a/src/viur/core/render/html/user.py b/src/viur/core/render/html/user.py index cdc7a4977..e8ef862dc 100644 --- a/src/viur/core/render/html/user.py +++ b/src/viur/core/render/html/user.py @@ -1,4 +1,4 @@ -from typing import Iterable +import typing as t from . import default as DefaultRender from viur.core.modules.user import UserSecondFactorAuthentication @@ -79,7 +79,7 @@ def second_factor_add_success(self, action_name: str, name: str, tpl: str | None return template.render(action_name=action_name, name=name) def second_factor_choice(self, - second_factors: Iterable[UserSecondFactorAuthentication], + second_factors: t.Iterable[UserSecondFactorAuthentication], tpl: str | None = None): tpl = self._choose_template(tpl, "second_factor_choice_template") template = self.getEnv().get_template(self.getTemplateFileName(tpl)) diff --git a/src/viur/core/render/html/utils.py b/src/viur/core/render/html/utils.py index 28a809f1f..86fb2abfa 100644 --- a/src/viur/core/render/html/utils.py +++ b/src/viur/core/render/html/utils.py @@ -1,4 +1,4 @@ -from typing import Callable, Union +import typing as t __jinjaGlobals_ = {} __jinjaFilters_ = {} @@ -38,7 +38,7 @@ def jinjaGlobalFilter(f): return f -def jinjaGlobalTest(func_or_alias: Union[Callable, str]) -> Callable: +def jinjaGlobalTest(func_or_alias: t.Callable | str) -> t.Callable: """ Decorator, marks a function as a Jinja2 test. diff --git a/src/viur/core/render/json/default.py b/src/viur/core/render/json/default.py index e0b9ce817..9ca9e4940 100644 --- a/src/viur/core/render/json/default.py +++ b/src/viur/core/render/json/default.py @@ -6,7 +6,7 @@ from viur.core.i18n import translate from viur.core.config import conf from datetime import datetime -from typing import Any, Dict, List, Optional, Tuple, Union +import typing as t class CustomJsonEncoder(json.JSONEncoder): @@ -14,7 +14,7 @@ class CustomJsonEncoder(json.JSONEncoder): This custom JSON-Encoder for this json-render ensures that translations are evaluated and can be dumped. """ - def default(self, o: Any) -> Any: + def default(self, o: t.Any) -> t.Any: if isinstance(o, translate): return str(o) elif isinstance(o, datetime): @@ -68,12 +68,11 @@ def render_structure(structure: dict): return structure - - def renderSingleBoneValue(self, value: Any, + def renderSingleBoneValue(self, value: t.Any, bone: bones.BaseBone, skel: SkeletonInstance, key - ) -> Union[Dict, str, None]: + ) -> dict | str | None: """ Renders the value of a bone. @@ -99,7 +98,7 @@ def renderSingleBoneValue(self, value: Any, return value return None - def renderBoneValue(self, bone: bones.BaseBone, skel: SkeletonInstance, key: str) -> Union[List, Dict, None]: + def renderBoneValue(self, bone: bones.BaseBone, skel: SkeletonInstance, key: str) -> list | dict | None: boneVal = skel[key] if bone.languages and bone.multiple: res = {} @@ -121,7 +120,7 @@ def renderBoneValue(self, bone: bones.BaseBone, skel: SkeletonInstance, key: str res = self.renderSingleBoneValue(boneVal, bone, skel, key) return res - def renderSkelValues(self, skel: SkeletonInstance, injectDownloadURL: bool = False) -> Optional[Dict]: + def renderSkelValues(self, skel: SkeletonInstance, injectDownloadURL: bool = False) -> t.Optional[dict]: """ Prepares values of one :class:`viur.core.skeleton.Skeleton` or a list of skeletons for output. diff --git a/src/viur/core/render/xml/default.py b/src/viur/core/render/xml/default.py index a1f44a755..1a5c323dd 100644 --- a/src/viur/core/render/xml/default.py +++ b/src/viur/core/render/xml/default.py @@ -1,4 +1,4 @@ -from typing import Any, Dict +import typing as t from viur.core.bones import * from viur.core import db @@ -89,7 +89,7 @@ def renderBoneValue(self, bone, skel, key): res = self.renderSingleBoneValue(boneVal, bone, skel, key) return res - def renderSingleBoneValue(self, value: Any, bone: baseBone, skel: SkeletonInstance, key: str) -> Dict: + def renderSingleBoneValue(self, value: t.Any, bone: baseBone, skel: SkeletonInstance, key: str) -> dict: """ Renders the value of a bone. diff --git a/src/viur/core/render/xml/user.py b/src/viur/core/render/xml/user.py index 246c72681..757016687 100644 --- a/src/viur/core/render/xml/user.py +++ b/src/viur/core/render/xml/user.py @@ -1,5 +1,3 @@ -import time, json -from string import Template from viur.core.render.xml.default import DefaultRender, serializeXML diff --git a/src/viur/core/request.py b/src/viur/core/request.py index a441a08f0..d7780ebe3 100644 --- a/src/viur/core/request.py +++ b/src/viur/core/request.py @@ -11,7 +11,7 @@ import os import time import traceback -import typing +import typing as t import inspect import unicodedata import webob @@ -40,7 +40,7 @@ class RequestValidator(ABC): @staticmethod @abstractmethod - def validate(request: 'BrowseHandler') -> typing.Optional[typing.Tuple[int, str, str]]: + def validate(request: 'BrowseHandler') -> t.Optional[tuple[int, str, str]]: """ The function that checks the current request. If the request is valid, simply return None. If the request should be blocked, it must return a tuple of @@ -61,7 +61,7 @@ class FetchMetaDataValidator(RequestValidator): name = "FetchMetaDataValidator" @staticmethod - def validate(request: 'BrowseHandler') -> typing.Optional[typing.Tuple[int, str, str]]: + def validate(request: 'BrowseHandler') -> t.Optional[tuple[int, str, str]]: headers = request.request.headers if not headers.get("sec-fetch-site"): # These headers are not send by all browsers return None diff --git a/src/viur/core/securityheaders.py b/src/viur/core/securityheaders.py index 5200b4e63..66ac0023e 100644 --- a/src/viur/core/securityheaders.py +++ b/src/viur/core/securityheaders.py @@ -42,7 +42,7 @@ from viur.core.config import conf from viur.core import current import logging -from typing import Literal, Optional, List +import typing as t def addCspRule(objectType: str, srcOrDirective: str, enforceMode: str = "monitor"): @@ -189,7 +189,7 @@ def enableStrictTransportSecurity(maxAge: int = 365 * 24 * 60 * 60, conf.security.strict_transport_security += "; preload" -def setXFrameOptions(action: str, uri: Optional[str] = None) -> None: +def setXFrameOptions(action: str, uri: t.Optional[str] = None) -> None: """ Sets X-Frame-Options to prevent click-jacking attacks. :param action: off | deny | sameorigin | allow-from @@ -205,7 +205,7 @@ def setXFrameOptions(action: str, uri: Optional[str] = None) -> None: conf.security.x_frame_options = (action, uri) -def setXXssProtection(enable: Optional[bool]) -> None: +def setXXssProtection(enable: t.Optional[bool]) -> None: """ Sets X-XSS-Protection header. If set, mode will always be block. :param enable: Enable the protection or not. Set to None to drop this header @@ -265,7 +265,7 @@ def _rebuildPermissionHeaderCache() -> None: ]) -def setPermissionPolicyDirective(directive: str, allowList: Optional[List[str]]) -> None: +def setPermissionPolicyDirective(directive: str, allowList: t.Optional[list[str]]) -> None: """ Set the permission policy. :param directive: The directive to set. diff --git a/src/viur/core/securitykey.py b/src/viur/core/securitykey.py index 882c31486..9b0400384 100644 --- a/src/viur/core/securitykey.py +++ b/src/viur/core/securitykey.py @@ -17,23 +17,23 @@ Therefor that header is prefixed with "Sec-" - so it cannot be read or set using JavaScript. """ -import typing +import typing as t import datetime import hmac from viur.core import conf, utils, current, db, tasks SECURITYKEY_KINDNAME = "viur-securitykey" SECURITYKEY_DURATION = 24 * 60 * 60 # one day -SECURITYKEY_STATIC_HEADER: typing.Final[str] = "Sec-X-ViUR-StaticSessionKey" +SECURITYKEY_STATIC_HEADER: t.Final[str] = "Sec-X-ViUR-StaticSessionKey" """The name of the header in which the static session key is provided at login and must be specified in requests that require a skey.""" -SECURITYKEY_STATIC_SKEY: typing.Final[str] = "STATIC_SESSION_KEY" +SECURITYKEY_STATIC_SKEY: t.Final[str] = "STATIC_SESSION_KEY" """Value that must be used as a marker in the payload (key: skey) to indicate that the session key from the headers should be used.""" def create( - duration: typing.Union[None, int] = None, + duration: None | int = None, session_bound: bool = True, key_length: int = 13, indexed: bool = True, @@ -73,7 +73,7 @@ def create( return key -def validate(key: str, session_bound: bool = True) -> typing.Union[bool, db.Entity]: +def validate(key: str, session_bound: bool = True) -> bool | db.Entity: """ Validates a CSRF-security-key. diff --git a/src/viur/core/session.py b/src/viur/core/session.py index cba654855..4b7002e3c 100644 --- a/src/viur/core/session.py +++ b/src/viur/core/session.py @@ -4,7 +4,7 @@ from viur.core.tasks import DeleteEntitiesIter from viur.core.config import conf # this import has to stay alone due partial import from viur.core import db, utils, tasks -from typing import Any, Optional, Union +import typing as t """ Provides the session implementation for the Google AppEngine™ based on the datastore. @@ -140,7 +140,7 @@ def __delitem__(self, key: str) -> None: del self.session[key] self.changed = True - def __getitem__(self, key) -> Any: + def __getitem__(self, key) -> t.Any: """ Returns the value stored under the given *key*. @@ -155,7 +155,7 @@ def __ior__(self, other: dict): self.session |= other return self - def get(self, key: str, default: Any = None) -> Any: + def get(self, key: str, default: t.Any = None) -> t.Any: """ Returns the value stored under the given key. @@ -164,7 +164,7 @@ def get(self, key: str, default: Any = None) -> Any: """ return self.session.get(key, default) - def __setitem__(self, key: str, item: Any): + def __setitem__(self, key: str, item: t.Any): """ Stores a new value under the given key. @@ -209,7 +209,7 @@ def items(self) -> 'dict_items': @tasks.CallDeferred -def killSessionByUser(user: Optional[Union[str, db.Key]] = None): +def killSessionByUser(user: t.Optional[t.Union[str, "db.Key", None]] = None): """ Invalidates all active sessions for the given *user*. diff --git a/src/viur/core/skeleton.py b/src/viur/core/skeleton.py index 4a5f8d296..ba14d4b07 100644 --- a/src/viur/core/skeleton.py +++ b/src/viur/core/skeleton.py @@ -5,12 +5,13 @@ import logging import os import string +import typing as t import sys import warnings from functools import partial from itertools import chain from time import time -from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Type, Union + from viur.core import conf, current, db, email, errors, translate, utils from viur.core.bones import BaseBone, DateBone, KeyBone, RelationalBone, RelationalUpdateLevel, SelectBone, StringBone @@ -98,7 +99,7 @@ def generate_bonemap(cls): return map -def skeletonByKind(kindName: str) -> Type[Skeleton]: +def skeletonByKind(kindName: str) -> t.Type[Skeleton]: """ Returns the Skeleton-Class for the given kindName. That skeleton must exist, otherwise an exception is raised. :param kindName: The kindname to retreive the skeleton for @@ -108,14 +109,14 @@ def skeletonByKind(kindName: str) -> Type[Skeleton]: return MetaBaseSkel._skelCache[kindName] -def listKnownSkeletons() -> List[str]: +def listKnownSkeletons() -> list[str]: """ :return: A list of all known kindnames (all kindnames for which a skeleton is defined) """ return list(MetaBaseSkel._skelCache.keys())[:] -def iterAllSkelClasses() -> Iterable["Skeleton"]: +def iterAllSkelClasses() -> t.Iterable["Skeleton"]: """ :return: An iterator that yields each Skeleton-Class once. (Only top-level skeletons are returned, so no RefSkel classes will be included) @@ -159,20 +160,20 @@ def __init__(self, skelCls, subSkelNames=None, fullClone=False, clonedBoneMap=No self.skeletonCls = skelCls self.renderPreparation = None - def items(self, yieldBoneValues: bool = False) -> Iterable[Tuple[str, BaseBone]]: + def items(self, yieldBoneValues: bool = False) -> t.Iterable[tuple[str, BaseBone]]: if yieldBoneValues: for key in self.boneMap.keys(): yield key, self[key] else: yield from self.boneMap.items() - def keys(self) -> Iterable[str]: + def keys(self) -> t.Iterable[str]: yield from self.boneMap.keys() - def values(self) -> Iterable[Any]: + def values(self) -> t.Iterable[t.Any]: yield from self.boneMap.values() - def __iter__(self) -> Iterable[str]: + def __iter__(self) -> t.Iterable[str]: yield from self.keys() def __contains__(self, item): @@ -362,8 +363,8 @@ def setSystemInitialized(cls): bone.setSystemInitialized() @classmethod - def setBoneValue(cls, skelValues: Any, boneName: str, value: Any, - append: bool = False, language: Optional[str] = None) -> bool: + def setBoneValue(cls, skelValues: t.Any, boneName: str, value: t.Any, + append: bool = False, language: t.Optional[str] = None) -> bool: """ Allow setting a bones value without calling fromClient or assigning to valuesCache directly. Santy-Checks are performed; if the value is invalid, that bone flips back to its original @@ -383,7 +384,7 @@ def setBoneValue(cls, skelValues: Any, boneName: str, value: Any, return bone.setBoneValue(skelValues, boneName, value, append, language) @classmethod - def fromClient(cls, skelValues: SkeletonInstance, data: Dict[str, Union[List[str], str]], + def fromClient(cls, skelValues: SkeletonInstance, data: dict[str, list[str] | str], allowEmptyRequired=False) -> bool: """ Load supplied *data* into Skeleton. @@ -523,7 +524,7 @@ class CustomDatabaseAdapter: # Indicate that we can run more types of queries than originally supported by firestore providesCustomQueries: bool = False - def preprocessEntry(self, entry: db.Entity, skel: BaseSkeleton, changeList: List[str], isAdd: bool) -> db.Entity: + def preprocessEntry(self, entry: db.Entity, skel: BaseSkeleton, changeList: list[str], isAdd: bool) -> db.Entity: """ Can be overridden to add or alter the data of this entry before it's written to firestore. Will always be called inside an transaction. @@ -535,7 +536,7 @@ def preprocessEntry(self, entry: db.Entity, skel: BaseSkeleton, changeList: List """ return entry - def updateEntry(self, dbObj: db.Entity, skel: BaseSkeleton, changeList: List[str], isAdd: bool) -> None: + def updateEntry(self, dbObj: db.Entity, skel: BaseSkeleton, changeList: list[str], isAdd: bool) -> None: """ Like `meth:preprocessEntry`, but runs after the transaction had completed. Changes made to dbObj will be ignored. @@ -554,7 +555,7 @@ def deleteEntry(self, entry: db.Entity, skel: BaseSkeleton) -> None: """ return - def fulltextSearch(self, queryString: str, databaseQuery: db.Query) -> List[db.Entity]: + def fulltextSearch(self, queryString: str, databaseQuery: db.Query) -> list[db.Entity]: """ If this database supports fulltext searches, this method has to implement them. If it's a plain fulltext search engine, leave 'prop:fulltextSearchGuaranteesQueryConstrains' set to False, @@ -593,7 +594,7 @@ def __init__(self, min_length: int = 3, max_length: int = 99, substring_matching self.max_length = max_length self.substring_matching = substring_matching - def _tagsFromString(self, value: str) -> Set[str]: + def _tagsFromString(self, value: str) -> set[str]: """ Extract all words including all min_length postfixes from given string """ @@ -611,7 +612,7 @@ def _tagsFromString(self, value: str) -> Set[str]: return res - def preprocessEntry(self, entry: db.Entity, skel: Skeleton, changeList: List[str], isAdd: bool) -> db.Entity: + def preprocessEntry(self, entry: db.Entity, skel: Skeleton, changeList: list[str], isAdd: bool) -> db.Entity: """ Collect searchTags from skeleton and build viurTags """ @@ -624,7 +625,7 @@ def preprocessEntry(self, entry: db.Entity, skel: Skeleton, changeList: List[str entry["viurTags"] = list(chain(*[self._tagsFromString(x) for x in tags if len(x) <= self.max_length])) return entry - def fulltextSearch(self, queryString: str, databaseQuery: db.Query) -> List[db.Entity]: + def fulltextSearch(self, queryString: str, databaseQuery: db.Query) -> list[db.Entity]: """ Run a fulltext search """ @@ -679,10 +680,10 @@ def serialize(self, skel: 'SkeletonInstance', name: str, parentIndexed: bool) -> class Skeleton(BaseSkeleton, metaclass=MetaSkel): kindName: str = _undefined # To which kind we save our data to - customDatabaseAdapter: Union[CustomDatabaseAdapter, None] = _undefined + customDatabaseAdapter: CustomDatabaseAdapter | None = _undefined subSkels = {} # List of pre-defined sub-skeletons of this type - interBoneValidations: List[ - Callable[[Skeleton], List[ReadFromClientError]]] = [] # List of functions checking inter-bone dependencies + interBoneValidations: list[ + t.Callable[[Skeleton], list[ReadFromClientError]]] = [] # List of functions checking inter-bone dependencies __seo_key_trans = str.maketrans( {"<": "", @@ -760,7 +761,7 @@ def all(cls, skelValues, **kwargs) -> db.Query: return db.Query(skelValues.kindName, srcSkelClass=skelValues, **kwargs) @classmethod - def fromClient(cls, skelValues: SkeletonInstance, data: Dict[str, Union[List[str], str]], + def fromClient(cls, skelValues: SkeletonInstance, data: dict[str, list[str] | str], allowEmptyRequired=False) -> bool: """ This function works similar to :func:`~viur.core.skeleton.Skeleton.setValues`, except that @@ -808,7 +809,7 @@ def fromClient(cls, skelValues: SkeletonInstance, data: Dict[str, Union[List[str return complete @classmethod - def fromDB(cls, skelValues: SkeletonInstance, key: Union[str, db.Key]) -> bool: + def fromDB(cls, skelValues: SkeletonInstance, key: str | db.Key) -> bool: """ Load entity with *key* from the data store into the Skeleton. @@ -1143,7 +1144,7 @@ def postDeletedHandler(cls, skelValues, key): pass @classmethod - def getCurrentSEOKeys(cls, skelValues) -> Union[None, Dict[str, str]]: + def getCurrentSEOKeys(cls, skelValues) -> None | dict[str, str]: """ Should be overridden to return a dictionary of language -> SEO-Friendly key this entry should be reachable under. How theses names are derived are entirely up to the application. @@ -1235,7 +1236,7 @@ class RelSkel(BaseSkeleton): """ @classmethod - def fromClient(cls, skelValues: SkeletonInstance, data: Dict[str, Union[List[str], str]], + def fromClient(cls, skelValues: SkeletonInstance, data: dict[str, list[str] | str], allowEmptyRequired=False) -> bool: """ Reads the data supplied by data. @@ -1294,7 +1295,7 @@ def serialize(self, parentIndexed): # return {k: v for k, v in self.valuesCache.entity.items() if k in self.__boneNames__} - def unserialize(self, values: Union[db.Entity, dict]): + def unserialize(self, values: db.Entity | dict): """ Loads 'values' into this skeleton. @@ -1328,7 +1329,7 @@ def unserialize(self, values: Union[db.Entity, dict]): class RefSkel(RelSkel): @classmethod - def fromSkel(cls, kindName: str, *args: List[str]) -> Type[RefSkel]: + def fromSkel(cls, kindName: str, *args: list[str]) -> t.Type[RefSkel]: """ Creates a relSkel from a skeleton-class using only the bones explicitly named in \*args @@ -1406,7 +1407,7 @@ def processRemovedRelations(removedKey, cursor=None): @CallDeferred -def updateRelations(destKey: db.Key, minChangeTime: int, changedBone: Optional[str], cursor: Optional[str] = None): +def updateRelations(destKey: db.Key, minChangeTime: int, changedBone: t.Optional[str], cursor: t.Optional[str] = None): """ This function updates Entities, which may have a copy of values from another entity which has been recently edited (updated). In ViUR, relations are implemented by copying the values from the referenced entity into the @@ -1506,12 +1507,12 @@ def _run(module: str, notify: str): class RebuildSearchIndex(QueryIter): @classmethod - def handleEntry(cls, skel: SkeletonInstance, customData: Dict[str, str]): + def handleEntry(cls, skel: SkeletonInstance, customData: dict[str, str]): skel.refresh() skel.toDB(update_relations=False) @classmethod - def handleFinish(cls, totalCount: int, customData: Dict[str, str]): + def handleFinish(cls, totalCount: int, customData: dict[str, str]): QueryIter.handleFinish(totalCount, customData) if not customData["notify"]: return diff --git a/src/viur/core/tasks.py b/src/viur/core/tasks.py index 620b1669a..cd5454a7c 100644 --- a/src/viur/core/tasks.py +++ b/src/viur/core/tasks.py @@ -8,11 +8,11 @@ import traceback from datetime import datetime, timedelta from functools import wraps +import typing as t from google.cloud import tasks_v2 from google.cloud.tasks_v2.services.cloud_tasks.transports import CloudTasksGrpcTransport from google.protobuf import timestamp_pb2 from time import sleep -from typing import Any, Callable, Dict, Optional, Tuple import json from viur.core import current, db, errors, utils @@ -95,7 +95,7 @@ def jsonDecodeObjectHook(obj): ) queueRegion = "local" -_periodicTasks: Dict[str, Dict[Callable, int]] = {} +_periodicTasks: dict[str, dict[t.Callable, int]] = {} _callableTasks = {} _deferred_tasks = {} _startupTasks = [] @@ -107,7 +107,7 @@ class PermanentTaskFailure(Exception): pass -def removePeriodicTask(task: Callable) -> None: +def removePeriodicTask(task: t.Callable) -> None: """ Removes a periodic task from the queue. Useful to unqueue an task that has been inherited from an overridden module. @@ -159,7 +159,8 @@ class TaskHandler(Module): adminInfo = None retryCountWarningThreshold = 25 - def findBoundTask(self, task: Callable, obj: object, depth: int = 0) -> Optional[Tuple[Callable, object]]: + def findBoundTask(self, task: t.Callable, obj: object, depth: int = 0) -> t.Optional[tuple[t.Callable, object]]: + """ Tries to locate the instance, this function belongs to. If it succeeds in finding it, it returns the function and its instance (-> its "self"). @@ -387,7 +388,7 @@ def execute(self, taskID, *args, **kwargs): ## Decorators ## def retry_n_times(retries: int, email_recipients: None | str | list[str] = None, - tpl: None | str = None) -> Callable: + tpl: None | str = None) -> t.Callable: """ Wrapper for deferred tasks to limit the amount of retries @@ -460,7 +461,7 @@ def noRetry(f): return retry_n_times(0)(f) -def CallDeferred(func: Callable) -> Callable: +def CallDeferred(func: t.Callable) -> t.Callable: """ This is a decorator, which always calls the wrapped method deferred. @@ -639,7 +640,7 @@ def callDeferred(func): return CallDeferred(func) -def PeriodicTask(interval: int = 0, cronName: str = "default") -> Callable: +def PeriodicTask(interval: int = 0, cronName: str = "default") -> t.Callable: """ Decorator to call a function periodic during maintenance. Interval defines a lower bound for the call-frequency for this task; @@ -663,7 +664,7 @@ def mkDecorator(fn): return mkDecorator -def CallableTask(fn: Callable) -> Callable: +def CallableTask(fn: t.Callable) -> t.Callable: """Marks a Class as representing a user-callable Task. It *should* extend CallableTaskBase and *must* provide its API @@ -673,7 +674,7 @@ def CallableTask(fn: Callable) -> Callable: return fn -def StartupTask(fn: Callable) -> Callable: +def StartupTask(fn: t.Callable) -> t.Callable: """ Functions decorated with this are called shortly at instance startup. It's *not* guaranteed that they actually run on the instance that just started up! @@ -721,7 +722,7 @@ class QueryIter(object, metaclass=MetaQueryIter): queueName = "default" # Name of the taskqueue we will run on @classmethod - def startIterOnQuery(cls, query: db.Query, customData: Any = None) -> None: + def startIterOnQuery(cls, query: db.Query, customData: t.Any = None) -> None: """ Starts iterating the given query on this class. Will return immediately, the first batch will already run deferred. @@ -748,7 +749,7 @@ def startIterOnQuery(cls, query: db.Query, customData: Any = None) -> None: cls._requeueStep(qryDict) @classmethod - def _requeueStep(cls, qryDict: Dict[str, Any]) -> None: + def _requeueStep(cls, qryDict: dict[str, t.Any]) -> None: """ Internal use only. Pushes a new step defined in qryDict to either the taskqueue or append it to the current request if we are on the local development server. @@ -774,7 +775,7 @@ def _requeueStep(cls, qryDict: Dict[str, Any]) -> None: )) @classmethod - def _qryStep(cls, qryDict: Dict[str, Any]) -> None: + def _qryStep(cls, qryDict: dict[str, t.Any]) -> None: """ Internal use only. Processes one block of five entries from the query defined in qryDict and reschedules the next block. diff --git a/src/viur/core/utils/__init__.py b/src/viur/core/utils/__init__.py index f049be1d1..13835ed2a 100644 --- a/src/viur/core/utils/__init__.py +++ b/src/viur/core/utils/__init__.py @@ -1,10 +1,11 @@ import hashlib import hmac import warnings + import logging from base64 import urlsafe_b64encode from datetime import datetime, timedelta, timezone -from typing import Any, Union, Optional +import typing as t from urllib.parse import quote from viur.core import current, db from viur.core.config import conf @@ -15,7 +16,7 @@ def utcNow() -> datetime: return datetime.now(timezone.utc) -def getCurrentUser() -> Optional["SkeletonInstance"]: +def getCurrentUser() -> t.Optional["SkeletonInstance"]: """ Retrieve current user, if logged in. If a user is logged in, this function returns a dict containing user data. @@ -52,14 +53,14 @@ def markFileForDeletion(dlkey: str) -> None: db.Put(fileObj) -def hmacSign(data: Any) -> str: +def hmacSign(data: t.Any) -> str: assert conf.file_hmac_key is not None, "No hmac-key set!" if not isinstance(data, bytes): data = str(data).encode("UTF-8") return hmac.new(conf.file_hmac_key, msg=data, digestmod=hashlib.sha3_384).hexdigest() -def hmacVerify(data: Any, signature: str) -> bool: +def hmacVerify(data: t.Any, signature: str) -> bool: return hmac.compare_digest(hmacSign(data), signature) @@ -75,8 +76,8 @@ def sanitizeFileName(fileName: str) -> str: def downloadUrlFor(folder: str, fileName: str, derived: bool = False, - expires: Union[timedelta, None] = timedelta(hours=1), - downloadFileName: Optional[str] = None) -> str: + expires: timedelta | None = timedelta(hours=1), + downloadFileName: t.Optional[str] = None) -> str: """ Utility function that creates a signed download-url for the given folder/filename combination @@ -106,7 +107,8 @@ def downloadUrlFor(folder: str, fileName: str, derived: bool = False, return "/file/download/%s?sig=%s" % (sigStr.decode("ASCII"), resstr) -def srcSetFor(fileObj: dict, expires: Optional[int], width: Optional[int] = None, height: Optional[int] = None) -> str: +def srcSetFor(fileObj: dict, expires: t.Optional[int], width: t.Optional[int] = None, + height: t.Optional[int] = None) -> str: """ Generates a string suitable for use as the srcset tag in html. This functionality provides the browser with a list of images in different sizes and allows it to choose the smallest file that will fill it's viewport @@ -147,9 +149,9 @@ def srcSetFor(fileObj: dict, expires: Optional[int], width: Optional[int] = None def seoUrlToEntry(module: str, - entry: Optional["SkeletonInstance"] = None, - skelType: Optional[str] = None, - language: Optional[str] = None) -> str: + entry: t.Optional["SkeletonInstance"] = None, + skelType: t.Optional[str] = None, + language: t.Optional[str] = None) -> str: """ Return the seo-url to a skeleton instance or the module. @@ -191,7 +193,7 @@ def seoUrlToEntry(module: str, return "/".join(pathComponents) -def seoUrlToFunction(module: str, function: str, render: Optional[str] = None) -> str: +def seoUrlToFunction(module: str, function: str, render: t.Optional[str] = None) -> str: from viur.core import conf lang = current.language.get() if module in conf.i18n.language_module_map and lang in conf.i18n.language_module_map[module]: @@ -216,7 +218,7 @@ def seoUrlToFunction(module: str, function: str, render: Optional[str] = None) - return "/".join(pathComponents) -def normalizeKey(key: Union[None, 'db.KeyClass']) -> Union[None, 'db.KeyClass']: +def normalizeKey(key: t.Union[None, 'db.KeyClass']) -> t.Union[None, 'db.KeyClass']: """ Normalizes a datastore key (replacing _application with the current one)