chore: BI-0 fix & enforce UP006 check

Author: KonstantAnxiety

lib/dl_api_client/dl_api_client/dsmaker/api/base.py

@@ -2,8 +2,6 @@
 
 from typing import (
     Any,
-    Dict,
-    List,
     Optional,
 )
 
@@ -22,7 +20,7 @@ class HttpApiResponse(ApiResponse):
     """
 
     _status_code: int = attr.ib(kw_only=True)
-    _json: Dict[str, Any] = attr.ib(kw_only=True)
+    _json: dict[str, Any] = attr.ib(kw_only=True)
 
     @property
     def status_code(self) -> int:
@@ -37,11 +35,11 @@ def json(self) -> dict:
         return self._json
 
     @property
-    def response_errors(self) -> List[str]:
+    def response_errors(self) -> list[str]:
         return self.extract_response_errors(self._json)
 
     @classmethod
-    def extract_response_errors(cls, response_json: Dict[str, Any]) -> List[str]:
+    def extract_response_errors(cls, response_json: dict[str, Any]) -> list[str]:
         return []
 
 

lib/dl_api_client/dl_api_client/dsmaker/api/data_api.py

@@ -3,7 +3,6 @@
 from typing import (
     TYPE_CHECKING,
     Any,
-    Dict,
     Mapping,
 )
 
@@ -191,12 +190,12 @@ class ValueSchema(OneOfSchemaWithDumpLoadHooks):
     }
 
     @pre_load(pass_many=False)
-    def wrap_value_with_type(self, data: Any, **_: Any) -> Dict[str, Any]:
+    def wrap_value_with_type(self, data: Any, **_: Any) -> dict[str, Any]:
         type = getattr(self, "context", {}).get(self.CONTEXT_KEY)
         return {"type": type, "value": data}
 
     @post_dump(pass_many=False)
-    def extract_value(self, data: Dict[str, Any], **_: Any) -> Any:
+    def extract_value(self, data: dict[str, Any], **_: Any) -> Any:
         return data["value"]
 
     def get_obj_type(self, obj: ParameterValue) -> str:
@@ -482,7 +481,7 @@ class DatasetContentInternalSchema(DefaultSchema[Dataset]):
     load_preview_by_default = ma_fields.Boolean(allow_none=True, dump_default=True, load_default=True)
 
     @post_load
-    def validate_rls2(self, item: Dict[str, Any], *args: Any, **kwargs: Any) -> Dict[str, Any]:
+    def validate_rls2(self, item: dict[str, Any], *args: Any, **kwargs: Any) -> dict[str, Any]:
         for key, entries in item["rls2"].items():
             for entry in entries:
                 if entry.pattern_type == RLSPatternType.value and entry.allowed_value is None:

lib/dl_api_client/dl_api_client/dsmaker/api/schemas/dataset.py

@@ -2,7 +2,6 @@
 
 from functools import singledispatchmethod
 from typing import (
-    List,
     Union,
 )
 
@@ -194,7 +193,7 @@ def dump_dataset(self, item: Dataset) -> dict:
         dataset_data = DatasetContentInternalSchema().dump(stripped_dataset)
         return {"dataset": dataset_data}
 
-    def generate_implicit_updates(self, dataset: Dataset) -> List[UpdateAction]:
+    def generate_implicit_updates(self, dataset: Dataset) -> list[UpdateAction]:
         updates = []
         for dsrc in dataset.sources:
             if not dsrc.created_:
@@ -224,7 +223,7 @@ def _get_action_postfix(item: ApiProxyObject) -> str:
             ObligatoryFilter: "obligatory_filter",
         }[type(item)]
 
-    def dump_updates(self, updates: List[Union[UpdateAction, dict]] = None) -> List[dict]:  # type: ignore  # 2024-01-24 # TODO: Incompatible default for argument "updates" (default has type "None", argument has type "list[UpdateAction | dict[Any, Any]]")  [assignment]
+    def dump_updates(self, updates: list[Union[UpdateAction, dict]] = None) -> list[dict]:  # type: ignore  # 2024-01-24 # TODO: Incompatible default for argument "updates" (default has type "None", argument has type "list[UpdateAction | dict[Any, Any]]")  [assignment]
         result = []
         for update in updates or ():
             if isinstance(update, UpdateAction):

lib/dl_api_client/dl_api_client/dsmaker/api/serialization_base.py

@@ -10,12 +10,10 @@
     Any,
     ClassVar,
     Collection,
-    Dict,
     Generator,
     Generic,
     List,
     Optional,
-    Tuple,
     TypeVar,
     Union,
 )
@@ -116,9 +114,9 @@ class Container(Generic[_ITEM_TV]):
     """
 
     def __init__(self, data: Union[list, tuple, dict, "Container"] = None):  # type: ignore  # 2024-01-24 # TODO: Incompatible default for argument "data" (default has type "None", argument has type "list[Any] | tuple[Any, ...] | dict[Any, Any] | Container[Any]")  [assignment]
-        self._item_ids: List[str] = []  # order list of object IDs
-        self._items: Dict[str, _ITEM_TV] = {}  # objects by ID
-        self._id_by_alias: Dict[str, str] = {}  # alias -> ID
+        self._item_ids: list[str] = []  # order list of object IDs
+        self._items: dict[str, _ITEM_TV] = {}  # objects by ID
+        self._id_by_alias: dict[str, str] = {}  # alias -> ID
 
         if data:
             self.__iadd__(data)
@@ -162,7 +160,7 @@ def __getitem__(self, key: Union[str, int]) -> _ITEM_TV:
 
         raise TypeError(f"Invalid key type for container: {type(key)}")
 
-    def items(self) -> Generator[Tuple[str, _ITEM_TV], None, None]:
+    def items(self) -> Generator[tuple[str, _ITEM_TV], None, None]:
         """Just like ``dict.items()`` - iterate over key-value tuples."""
         alias_by_id = {id: alias for alias, id in self._id_by_alias.items()}
         for id in self._item_ids:
@@ -261,7 +259,7 @@ def set_name(self, name: str) -> None:
     def join(
         self,
         other: "SourceAvatar",
-        conditions: List["JoinCondition"] = None,  # type: ignore  # 2024-01-24 # TODO: Incompatible default for argument "conditions" (default has type "None", argument has type "list[JoinCondition]")  [assignment]
+        conditions: list["JoinCondition"] = None,  # type: ignore  # 2024-01-24 # TODO: Incompatible default for argument "conditions" (default has type "None", argument has type "list[JoinCondition]")  [assignment]
         join_type: JoinType = JoinType.inner,
     ) -> "AvatarRelation":
         return AvatarRelation(
@@ -479,7 +477,7 @@ class RangeParameterValueConstraint(BaseParameterValueConstraint):
 @attr.s
 class SetParameterValueConstraint(BaseParameterValueConstraint):
     type: ParameterValueConstraintType = ParameterValueConstraintType.set
-    values: List[ParameterValue] = attr.ib(factory=list)
+    values: list[ParameterValue] = attr.ib(factory=list)
 
 
 @attr.s
@@ -502,7 +500,7 @@ class RegexParameterValueConstraint(BaseParameterValueConstraint):
 
 class CollectionParameterValueConstraint(BaseParameterValueConstraint):
     type: ParameterValueConstraintType = ParameterValueConstraintType.collection
-    constraints: List[BaseParameterValueConstraint]
+    constraints: list[BaseParameterValueConstraint]
 
 
 def _make_pivot_role_spec(
@@ -611,7 +609,7 @@ def as_req_legend_item(
         self,
         role: FieldRole = FieldRole.row,
         range_type: Optional[RangeType] = None,
-        dimension_values: Optional[Dict[int, Any]] = None,
+        dimension_values: Optional[dict[int, Any]] = None,
         tree_prefix: Optional[list] = None,
         tree_level: Optional[int] = None,
         legend_item_id: Optional[int] = None,
@@ -674,7 +672,7 @@ def for_block(self, block_id: int) -> WhereClause:
 @attr.s
 class ObligatoryFilter(ApiProxyObject):
     field_guid: str = attr.ib(default="")
-    default_filters: List[WhereClause] = attr.ib(factory=list)
+    default_filters: list[WhereClause] = attr.ib(factory=list)
     managed_by: ManagedBy = attr.ib(default=ManagedBy.user)
     valid: bool = attr.ib(default=True)
 
@@ -708,7 +706,7 @@ class TemplateRoleSpec(DimensionRoleSpec):  # noqa
 class TreeRoleSpec(DimensionRoleSpec):
     level: int = attr.ib(kw_only=True)
     prefix: str = attr.ib(kw_only=True)
-    dimension_values: List[DimensionValueSpec] = attr.ib(kw_only=True, factory=list)
+    dimension_values: list[DimensionValueSpec] = attr.ib(kw_only=True, factory=list)
 
 
 @attr.s
@@ -787,7 +785,7 @@ class DimensionValueSpec:
 class AfterBlockPlacement(BlockPlacement):
     type = QueryBlockPlacementType.after
 
-    dimension_values: Optional[List[DimensionValueSpec]] = attr.ib(default=None)
+    dimension_values: Optional[list[DimensionValueSpec]] = attr.ib(default=None)
 
 
 @attr.s
@@ -914,12 +912,12 @@ class ComponentError:
 class ComponentErrorPack:
     id: str = attr.ib()
     type: ComponentType = attr.ib()
-    errors: List[ComponentError] = attr.ib(factory=list)
+    errors: list[ComponentError] = attr.ib(factory=list)
 
 
 @attr.s
 class ComponentErrorRegistry:
-    items: List[ComponentErrorPack] = attr.ib(factory=list)
+    items: list[ComponentErrorPack] = attr.ib(factory=list)
 
     def get_pack(self, id: str) -> Optional[ComponentErrorPack]:  # type: ignore  # TODO: fix
         for item in self.items:
@@ -943,9 +941,9 @@ class Dataset(ApiProxyObject):
     result_schema: Container[ResultField] = attr.ib(factory=Container, converter=Container)
     result_schema_aux: ResultSchemaAux = attr.ib(factory=ResultSchemaAux)
     rls: dict = attr.ib(factory=dict)
-    rls2: dict[str, List[RLSEntry]] = attr.ib(factory=dict)
+    rls2: dict[str, list[RLSEntry]] = attr.ib(factory=dict)
     component_errors: ComponentErrorRegistry = attr.ib(factory=ComponentErrorRegistry)
-    obligatory_filters: List[ObligatoryFilter] = attr.ib(default=attr.Factory(list))
+    obligatory_filters: list[ObligatoryFilter] = attr.ib(default=attr.Factory(list))
 
     def prepare(self) -> None:
         super().prepare()

lib/dl_api_client/dl_api_client/dsmaker/primitives.py

@@ -8,9 +8,7 @@
 import logging
 from typing import (
     Any,
-    Dict,
     Optional,
-    Tuple,
 )
 
 from aiohttp import web
@@ -34,7 +32,7 @@ class ErrorLevel(enum.Enum):
 @attr.s(auto_attribs=True)
 class ErrorData:
     status_code: int
-    response_body: Dict[str, Any]
+    response_body: dict[str, Any]
     level: ErrorLevel = attr.ib(validator=[attr.validators.instance_of(ErrorLevel)])
     http_reason: Optional[str] = attr.ib(default=None)
 
@@ -66,7 +64,7 @@ def _classify_error(self, err: Exception, request: web.Request) -> ErrorData:
 
     def handle_error(
         self, err: Exception, request: web.Request, req_logging_ctx_ctrl: RequestLoggingContextController
-    ) -> Tuple[web.Response, ErrorData]:
+    ) -> tuple[web.Response, ErrorData]:
         if isinstance(err, CancelledError):
             LOGGER.warning("Client request was cancelled", exc_info=True)
             raise  # noqa

lib/dl_api_commons/dl_api_commons/aio/middlewares/error_handling_outer.py

@@ -4,7 +4,6 @@
 import logging
 from typing import (
     Optional,
-    Type,
 )
 
 from aiohttp import web
@@ -40,7 +39,7 @@
 @attr.s
 class RequestId:
     header_name: str = attr.ib(default="X-Request-ID")
-    dl_request_cls: Type[aiohttp_wrappers.DLRequestBase] = attr.ib(default=aiohttp_wrappers.DLRequestBase)
+    dl_request_cls: type[aiohttp_wrappers.DLRequestBase] = attr.ib(default=aiohttp_wrappers.DLRequestBase)
     append_own_req_id: bool = attr.ib(default=False)
     app_prefix: Optional[str] = attr.ib(default=None)
     accept_logging_ctx: bool = attr.ib(default=False)

lib/dl_api_commons/dl_api_commons/aio/middlewares/request_id.py

@@ -15,7 +15,6 @@
     Awaitable,
     Callable,
     Optional,
-    Type,
 )
 
 import aiohttp
@@ -51,7 +50,7 @@ class PredefinedIntervalsRetrier(BaseRetrier):
     retry_intervals: tuple[float, ...] = attr.ib(default=(0.5, 1.0, 2.0))
     retry_codes: set[int] = attr.ib(default={408, 429, 500, 502, 503, 504})
     retry_methods: set[str] = attr.ib(default={"GET"})
-    retry_exception_classes: tuple[Type[Exception]] = attr.ib(default=(aiohttp.ClientError,))
+    retry_exception_classes: tuple[type[Exception]] = attr.ib(default=(aiohttp.ClientError,))
 
     @retry_methods.validator
     def check_all_chars_upper(self, attribute: str, value: set[str]) -> None:
@@ -127,7 +126,7 @@ async def __aenter__(self) -> BIAioHTTPClient:
         return self
 
     async def __aexit__(
-        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
+        self, exc_type: Optional[type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
     ) -> None:
         await self.close()
 

lib/dl_api_commons/dl_api_commons/aiohttp/aiohttp_client.py

@@ -8,11 +8,9 @@
     Any,
     Awaitable,
     Callable,
-    FrozenSet,
     Generic,
     Literal,
     Optional,
-    Type,
     TypeVar,
     Union,
     overload,
@@ -225,7 +223,7 @@ def get_single_json_header(self, header: DLHeaders) -> Union[bool, int, float, l
             ) from e
 
     @property
-    def required_resources(self) -> FrozenSet[RequiredResource]:
+    def required_resources(self) -> frozenset[RequiredResource]:
         handler = self.request.match_info.handler
 
         if hasattr(handler, self.HANDLER_ATTR_NAME_REQUIRED_RESOURCES):
@@ -237,7 +235,7 @@ def required_resources(self) -> FrozenSet[RequiredResource]:
 
     @classmethod
     def use_dl_request(
-        cls: Type[_SELF_TYPE], coro: Callable[[_SELF_TYPE, Handler], Awaitable[web.StreamResponse]]
+        cls: type[_SELF_TYPE], coro: Callable[[_SELF_TYPE, Handler], Awaitable[web.StreamResponse]]
     ) -> Middleware:
         if not inspect.iscoroutinefunction(coro):
             raise ValueError("This decorator may only be applied to a coroutine")
@@ -274,7 +272,7 @@ async def wrapper(self: Any, request: web.Request, handler: Handler) -> web.Stre
 
 
 class DLRequestView(web.View, Generic[_DL_REQUEST_TV]):
-    dl_request_cls: Type[_DL_REQUEST_TV] = DLRequestBase  # type: ignore  # 2024-01-30 # TODO: Incompatible types in assignment (expression has type "type[DLRequestBase]", variable has type "type[_DL_REQUEST_TV]")  [assignment]
+    dl_request_cls: type[_DL_REQUEST_TV] = DLRequestBase  # type: ignore  # 2024-01-30 # TODO: Incompatible types in assignment (expression has type "type[DLRequestBase]", variable has type "type[_DL_REQUEST_TV]")  [assignment]
 
     @property
     def dl_request(self) -> _DL_REQUEST_TV:

lib/dl_api_commons/dl_api_commons/aiohttp/aiohttp_wrappers.py

@@ -5,7 +5,6 @@
 from typing import (
     Any,
     Optional,
-    Type,
     TypeVar,
     Union,
 )
@@ -128,12 +127,12 @@ def normalize_headers_dict(headers: Union[CIMultiDict, dict, None]) -> CIMultiDi
         raise TypeError(f"Unexpected type of headers: {type(headers)}")
 
     @classmethod
-    def create_empty(cls: Type[_REQUEST_CONTEXT_INFO_TV]) -> _REQUEST_CONTEXT_INFO_TV:
+    def create_empty(cls: type[_REQUEST_CONTEXT_INFO_TV]) -> _REQUEST_CONTEXT_INFO_TV:
         return cls()
 
     @classmethod
     def create(
-        cls: Type[_REQUEST_CONTEXT_INFO_TV],
+        cls: type[_REQUEST_CONTEXT_INFO_TV],
         request_id: Optional[str],
         tenant: Optional[TenantDef],
         user_id: Optional[str],