Type annotations update

Author: HardNorth

reportportal_client/__init__.py

@@ -15,7 +15,7 @@
 
 import sys
 import warnings
-from typing import Optional, Tuple, TypedDict, Union
+from typing import Optional, TypedDict, Union
 
 # noinspection PyUnreachableCode
 if sys.version_info >= (3, 11):
@@ -60,7 +60,7 @@ class _ClientOptions(TypedDict, total=False):
     verify_ssl: Union[bool, str]
     retries: int
     max_pool_size: int
-    http_timeout: Union[float, Tuple[float, float]]
+    http_timeout: Union[float, tuple[float, float]]
     mode: str
     launch_uuid_print: bool
     print_output: OutputType

reportportal_client/_internal/aio/http.py

@@ -24,7 +24,7 @@
 import asyncio
 import sys
 from types import TracebackType
-from typing import Any, Callable, Coroutine, Optional, Type, Union
+from typing import Any, Callable, Coroutine, Optional, Union
 
 from aenum import Enum
 from aiohttp import ClientResponse, ClientResponseError
@@ -160,7 +160,7 @@ async def __aenter__(self) -> "RetryingClientSession":
 
     async def __aexit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc_val: Optional[BaseException],
         exc_tb: Optional[TracebackType],
     ) -> None:
@@ -241,7 +241,7 @@ async def __aenter__(self) -> "ClientSession":
 
     async def __aexit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc_val: Optional[BaseException],
         exc_tb: Optional[TracebackType],
     ) -> None:

reportportal_client/_internal/aio/tasks.py

@@ -17,7 +17,7 @@
 import sys
 import time
 from asyncio import Future
-from typing import Any, Awaitable, Coroutine, Generator, Generic, List, Optional, TypeVar, Union
+from typing import Any, Awaitable, Coroutine, Generator, Generic, Optional, TypeVar, Union
 
 from reportportal_client.aio.tasks import BlockingOperationError, Task
 
@@ -142,7 +142,7 @@ def __call__(
 class TriggerTaskBatcher(Generic[_T]):
     """Batching class which compile its batches by object number or by passed time."""
 
-    __task_list: List[_T]
+    __task_list: list[_T]
     __last_run_time: float
     __trigger_num: int
     __trigger_interval: float
@@ -170,7 +170,7 @@ def __ready_to_run(self) -> bool:
             return True
         return False
 
-    def append(self, value: _T) -> Optional[List[_T]]:
+    def append(self, value: _T) -> Optional[list[_T]]:
         """Add an object to internal batch and return the batch if it's triggered.
 
         :param   value: an object to add to the batch
@@ -184,7 +184,7 @@ def append(self, value: _T) -> Optional[List[_T]]:
         self.__task_list = []
         return tasks
 
-    def flush(self) -> Optional[List[_T]]:
+    def flush(self) -> Optional[list[_T]]:
         """Immediately return everything what's left in the internal batch.
 
         :return: a batch or None
@@ -200,7 +200,7 @@ def flush(self) -> Optional[List[_T]]:
 class BackgroundTaskList(Generic[_T]):
     """Task list class which collects Tasks into internal batch and removes when they complete."""
 
-    __task_list: List[_T]
+    __task_list: list[_T]
 
     def __init__(self):
         """Initialize an instance of the Batcher."""
@@ -222,7 +222,7 @@ def append(self, value: _T) -> None:
         self.__remove_finished()
         self.__task_list.append(value)
 
-    def flush(self) -> Optional[List[_T]]:
+    def flush(self) -> Optional[list[_T]]:
         """Immediately return everything what's left unfinished in the internal batch.
 
         :return: a batch or None

reportportal_client/_internal/http.py

@@ -15,7 +15,7 @@
 """This module designed to help with synchronous HTTP request/response handling."""
 
 from types import TracebackType
-from typing import Any, Callable, Optional, Type, Union
+from typing import Any, Callable, Optional, Union
 
 from requests import Response, Session
 from requests.adapters import BaseAdapter
@@ -104,7 +104,7 @@ def __enter__(self) -> "ClientSession":
 
     def __exit__(
         self,
-        exc_type: Optional[Type[BaseException]],
+        exc_type: Optional[type[BaseException]],
         exc_val: Optional[BaseException],
         exc_tb: Optional[TracebackType],
     ) -> None:

reportportal_client/_internal/logs/batcher.py

@@ -15,7 +15,7 @@
 
 import logging
 import threading
-from typing import Any, Dict, Generic, List, Optional, TypeVar
+from typing import Any, Generic, Optional, TypeVar
 
 from reportportal_client.core.rp_requests import AsyncRPRequestLog, RPRequestLog
 from reportportal_client.logs import MAX_LOG_BATCH_PAYLOAD_SIZE, MAX_LOG_BATCH_SIZE
@@ -35,7 +35,7 @@ class LogBatcher(Generic[T_co]):
     entry_num: int
     payload_limit: int
     _lock: threading.Lock
-    _batch: List[T_co]
+    _batch: list[T_co]
     _payload_size: int
 
     def __init__(self, entry_num=MAX_LOG_BATCH_SIZE, payload_limit=MAX_LOG_BATCH_PAYLOAD_SIZE) -> None:
@@ -50,7 +50,7 @@ def __init__(self, entry_num=MAX_LOG_BATCH_SIZE, payload_limit=MAX_LOG_BATCH_PAY
         self._batch = []
         self._payload_size = 0
 
-    def _append(self, size: int, log_req: RPRequestLog) -> Optional[List[RPRequestLog]]:
+    def _append(self, size: int, log_req: RPRequestLog) -> Optional[list[RPRequestLog]]:
         with self._lock:
             if self._payload_size + size >= self.payload_limit:
                 if len(self._batch) > 0:
@@ -68,23 +68,23 @@ def _append(self, size: int, log_req: RPRequestLog) -> Optional[List[RPRequestLo
             self._payload_size = 0
             return batch
 
-    def append(self, log_req: RPRequestLog) -> Optional[List[RPRequestLog]]:
+    def append(self, log_req: RPRequestLog) -> Optional[list[RPRequestLog]]:
         """Add a log request object to internal batch and return the batch if it's full.
 
         :param   log_req: log request object
         :return: a batch or None
         """
         return self._append(log_req.multipart_size, log_req)
 
-    async def append_async(self, log_req: AsyncRPRequestLog) -> Optional[List[AsyncRPRequestLog]]:
+    async def append_async(self, log_req: AsyncRPRequestLog) -> Optional[list[AsyncRPRequestLog]]:
         """Add a log request object to internal batch and return the batch if it's full.
 
         :param   log_req: log request object
         :return: a batch or None
         """
         return self._append(await log_req.multipart_size, log_req)
 
-    def flush(self) -> Optional[List[T_co]]:
+    def flush(self) -> Optional[list[T_co]]:
         """Immediately return everything what's left in the internal batch.
 
         :return: a batch or None
@@ -99,7 +99,7 @@ def flush(self) -> Optional[List[T_co]]:
             self._payload_size = 0
             return batch
 
-    def __getstate__(self) -> Dict[str, Any]:
+    def __getstate__(self) -> dict[str, Any]:
         """Control object pickling and return object fields as Dictionary.
 
         :return: object state dictionary
@@ -110,7 +110,7 @@ def __getstate__(self) -> Dict[str, Any]:
         del state["_lock"]
         return state
 
-    def __setstate__(self, state: Dict[str, Any]) -> None:
+    def __setstate__(self, state: dict[str, Any]) -> None:
         """Control object pickling, receives object state as Dictionary.
 
         :param dict state: object state dictionary

reportportal_client/_internal/services/statistics.py

@@ -16,7 +16,7 @@
 import logging
 import ssl
 from platform import python_version
-from typing import Optional, Tuple
+from typing import Optional
 
 import aiohttp
 import certifi
@@ -31,7 +31,7 @@
 ID, KEY = CLIENT_INFO.split(":")
 
 
-def _get_client_info() -> Tuple[str, str]:
+def _get_client_info() -> tuple[str, str]:
     """Get name of the client and its version.
 
     :return: ('reportportal-client', '5.0.4')

reportportal_client/aio/client.py

@@ -20,7 +20,7 @@
 import time as datetime
 import warnings
 from os import getenv
-from typing import Any, Coroutine, Dict, List, Optional, Tuple, TypeVar, Union
+from typing import Any, Coroutine, Optional, TypeVar, Union
 
 import aiohttp
 import certifi
@@ -115,7 +115,7 @@ class Client:
     verify_ssl: Union[bool, str]
     retries: Optional[int]
     max_pool_size: int
-    http_timeout: Optional[Union[float, Tuple[float, float]]]
+    http_timeout: Optional[Union[float, tuple[float, float]]]
     keepalive_timeout: Optional[float]
     mode: str
     launch_uuid_print: bool
@@ -135,7 +135,7 @@ def __init__(
         verify_ssl: Union[bool, str] = True,
         retries: int = NOT_SET,
         max_pool_size: int = 50,
-        http_timeout: Optional[Union[float, Tuple[float, float]]] = (10, 10),
+        http_timeout: Optional[Union[float, tuple[float, float]]] = (10, 10),
         keepalive_timeout: Optional[float] = None,
         mode: str = "DEFAULT",
         launch_uuid_print: bool = False,
@@ -259,12 +259,12 @@ async def session(self) -> ClientSession:
             else:
                 ssl_config = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=certifi.where())
 
-        connection_params: Dict[str, Any] = {"ssl": ssl_config, "limit": self.max_pool_size}
+        connection_params: dict[str, Any] = {"ssl": ssl_config, "limit": self.max_pool_size}
         if self.keepalive_timeout:
             connection_params["keepalive_timeout"] = self.keepalive_timeout
         connector = aiohttp.TCPConnector(**connection_params)
 
-        session_params: Dict[str, Any] = {"connector": connector}
+        session_params: dict[str, Any] = {"connector": connector}
 
         if self.http_timeout:
             if type(self.http_timeout) is tuple:
@@ -366,7 +366,7 @@ async def start_test_item(
         *,
         parent_item_id: Optional[Union[str, Task[str]]] = None,
         description: Optional[str] = None,
-        attributes: Optional[Union[List[dict], dict]] = None,
+        attributes: Optional[Union[list[dict], dict]] = None,
         parameters: Optional[dict] = None,
         code_ref: Optional[str] = None,
         test_case_id: Optional[str] = None,
@@ -629,7 +629,7 @@ async def get_project_settings(self) -> Optional[dict]:
         response = await AsyncHttpRequest((await self.session()).get, url=url, name="get_project_settings").make()
         return await response.json if response else None
 
-    async def log_batch(self, log_batch: Optional[List[AsyncRPRequestLog]]) -> Optional[Tuple[str, ...]]:
+    async def log_batch(self, log_batch: Optional[list[AsyncRPRequestLog]]) -> Optional[tuple[str, ...]]:
         """Send batch logging message to the ReportPortal.
 
         :param log_batch: A list of log message objects.
@@ -672,7 +672,7 @@ def clone(self) -> "Client":
         )
         return cloned
 
-    def __getstate__(self) -> Dict[str, Any]:
+    def __getstate__(self) -> dict[str, Any]:
         """Control object pickling and return object fields as Dictionary.
 
         :return: object state dictionary
@@ -683,7 +683,7 @@ def __getstate__(self) -> Dict[str, Any]:
         del state["_session"]
         return state
 
-    def __setstate__(self, state: Dict[str, Any]) -> None:
+    def __setstate__(self, state: dict[str, Any]) -> None:
         """Control object pickling, receives object state as Dictionary.
 
         :param dict state: object state dictionary
@@ -847,7 +847,7 @@ async def start_test_item(
         start_time: str,
         item_type: str,
         description: Optional[str] = None,
-        attributes: Optional[List[dict]] = None,
+        attributes: Optional[list[dict]] = None,
         parameters: Optional[dict] = None,
         parent_item_id: Optional[str] = None,
         has_stats: bool = True,
@@ -1047,7 +1047,7 @@ async def log(
         level: Optional[Union[int, str]] = None,
         attachment: Optional[dict] = None,
         item_id: Optional[str] = None,
-    ) -> Optional[Tuple[str, ...]]:
+    ) -> Optional[tuple[str, ...]]:
         """Send Log message to the ReportPortal and attach it to a Test Item or Launch.
 
         This method stores Log messages in internal batch and sent it when batch is full, so not every method
@@ -1293,7 +1293,7 @@ def start_test_item(
         start_time: str,
         item_type: str,
         description: Optional[str] = None,
-        attributes: Optional[List[dict]] = None,
+        attributes: Optional[list[dict]] = None,
         parameters: Optional[dict] = None,
         parent_item_id: Optional[Task[str]] = None,
         has_stats: bool = True,
@@ -1485,10 +1485,10 @@ def get_project_settings(self) -> Task[Optional[str]]:
         result_task = self.create_task(result_coro)
         return result_task
 
-    async def _log_batch(self, log_rq: Optional[List[AsyncRPRequestLog]]) -> Optional[Tuple[str, ...]]:
+    async def _log_batch(self, log_rq: Optional[list[AsyncRPRequestLog]]) -> Optional[tuple[str, ...]]:
         return await self.__client.log_batch(log_rq)
 
-    async def _log(self, log_rq: AsyncRPRequestLog) -> Optional[Tuple[str, ...]]:
+    async def _log(self, log_rq: AsyncRPRequestLog) -> Optional[tuple[str, ...]]:
         return await self._log_batch(await self._log_batcher.append_async(log_rq))
 
     def log(
@@ -1498,7 +1498,7 @@ def log(
         level: Optional[Union[int, str]] = None,
         attachment: Optional[dict] = None,
         item_id: Optional[Task[str]] = None,
-    ) -> Task[Optional[Tuple[str, ...]]]:
+    ) -> Task[Optional[tuple[str, ...]]]:
         """Send Log message to the ReportPortal and attach it to a Test Item or Launch.
 
         This method stores Log messages in internal batch and sent it when batch is full, so not every method
@@ -1697,7 +1697,7 @@ def clone(self) -> "ThreadedRPClient":
             cloned._add_current_item(current_item)
         return cloned
 
-    def __getstate__(self) -> Dict[str, Any]:
+    def __getstate__(self) -> dict[str, Any]:
         """Control object pickling and return object fields as Dictionary.
 
         :return: object state dictionary
@@ -1710,7 +1710,7 @@ def __getstate__(self) -> Dict[str, Any]:
         del state["_thread"]
         return state
 
-    def __setstate__(self, state: Dict[str, Any]) -> None:
+    def __setstate__(self, state: dict[str, Any]) -> None:
         """Control object pickling, receives object state as Dictionary.
 
         :param dict state: object state dictionary
@@ -1889,7 +1889,7 @@ def clone(self) -> "BatchedRPClient":
             cloned._add_current_item(current_item)
         return cloned
 
-    def __getstate__(self) -> Dict[str, Any]:
+    def __getstate__(self) -> dict[str, Any]:
         """Control object pickling and return object fields as Dictionary.
 
         :return: object state dictionary
@@ -1901,7 +1901,7 @@ def __getstate__(self) -> Dict[str, Any]:
         del state["_loop"]
         return state
 
-    def __setstate__(self, state: Dict[str, Any]) -> None:
+    def __setstate__(self, state: dict[str, Any]) -> None:
         """Control object pickling, receives object state as Dictionary.
 
         :param dict state: object state dictionary