remove object base class, replace typing aliases

examples/opentelemetry/load_tank.py

@@ -7,7 +7,7 @@
 import random
 import time
 from dataclasses import dataclass
-from typing import AsyncIterator, Tuple
+from typing import AsyncIterator
 
 from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
 from opentelemetry.sdk.metrics import MeterProvider
@@ -84,7 +84,7 @@ def _load_config() -> LoadConfig:
     )
 
 
-async def _load_steps(config: LoadConfig) -> AsyncIterator[Tuple[int, str, int]]:
+async def _load_steps(config: LoadConfig) -> AsyncIterator[tuple[int, str, int]]:
     pattern = (
         (config.peak_rps, "Peak", config.peak_duration),
         (config.medium_rps, "Medium down", config.medium_duration),

examples/reservations-bot-demo/cloud_function/controller.py

@@ -1,4 +1,3 @@
-import typing
 import logging
 from storage import Storage
 from models import (
@@ -16,7 +15,7 @@ class Controller(object):
     def __init__(self, storage: Storage):
         self._storage = storage
 
-    def _find_available_table_id(self, request: ReservationCreateRequest) -> typing.Optional[int]:
+    def _find_available_table_id(self, request: ReservationCreateRequest) -> int | None:
         table_ids = set(self._storage.list_table_ids(cnt=request.cnt))
         reserved_table_ids = set(self._storage.find_reserved_table_ids(cnt=request.cnt, dt=request.dt))
         for table_id in table_ids.difference(reserved_table_ids):

examples/reservations-bot-demo/cloud_function/models.py

@@ -1,35 +1,34 @@
-import typing
 import datetime
 from pydantic import BaseModel
 
 
 class Reservation(BaseModel):
-    phone: typing.Optional[typing.Union[str, str]] = None
-    description: typing.Optional[typing.Union[bytes, str]] = None
+    phone: str | str | None = None
+    description: bytes | str | None = None
     table_id: int
     dt: datetime.datetime
 
 
 class Table(BaseModel):
     table_id: int = None
-    description: typing.Optional[typing.Union[bytes, str]] = None
+    description: bytes | str | None = None
     cnt: int
 
 
 class ReservationCreateRequest(BaseModel):
     dt: datetime.datetime
     cnt: int
-    description: typing.Optional[typing.Union[bytes, str]] = None
-    phone: typing.Optional[typing.Union[bytes, str]]
+    description: bytes | str | None = None
+    phone: bytes | str | None
 
 
 class ReservationCreateResponse(BaseModel):
     success: bool
-    table_id: typing.Optional[int] = None
+    table_id: int | None = None
 
 
 class ReservationCancelRequest(BaseModel):
-    phone: typing.Optional[typing.Union[bytes, str]]
+    phone: bytes | str | None
     dt: datetime.datetime
 
 

examples/reservations-bot-demo/cloud_function/storage.py

@@ -1,5 +1,4 @@
 import datetime
-import typing
 import ydb
 from utils import session_pool_context, make_driver_config
 from config import Config
@@ -10,7 +9,7 @@ def __init__(self, *, endpoint: str, database: str, path: str):
         self._database = database
         self._driver_config = make_driver_config(endpoint, database, path)
 
-    def list_table_ids(self, *, cnt: int = 0) -> typing.List[int]:
+    def list_table_ids(self, *, cnt: int = 0) -> list[int]:
         query = f"""PRAGMA TablePathPrefix("{self._database}");
         DECLARE $cnt as Uint64;
         SELECT table_id FROM tables WHERE cnt >= $cnt;
@@ -26,7 +25,7 @@ def transaction(session):
             tables = session_pool.retry_operation_sync(transaction)
             return list(map(lambda x: getattr(x, "table_id"), tables))
 
-    def find_reserved_table_ids(self, *, cnt: int, dt: datetime.datetime) -> typing.List[int]:
+    def find_reserved_table_ids(self, *, cnt: int, dt: datetime.datetime) -> list[int]:
         query = f"""PRAGMA TablePathPrefix("{self._database}");
         DECLARE $dt AS DateTime;
         DECLARE $reservation_period_minutes AS Int32;

examples/time-series-serverless/database.py

@@ -1,5 +1,4 @@
 import ydb
-from typing import List
 
 from config import ydb_configuration
 from exception import ConnectionFailure
@@ -31,7 +30,7 @@ def create_driver(self) -> ydb.Driver:
     def table_client(self) -> ydb.TableClient:
         return self.driver.table_client
 
-    def bulk_upsert(self, rows: List, column_types: ydb.BulkUpsertColumns):
+    def bulk_upsert(self, rows: list, column_types: ydb.BulkUpsertColumns):
         self.table_client.bulk_upsert(self.config.full_path, rows, column_types)
 
 

examples/time-series-serverless/time_series.py

@@ -1,4 +1,3 @@
-from typing import Dict
 import random
 
 import ydb
@@ -23,7 +22,7 @@ def generate_time_series(parameters: Parameters):
     ydb_client.bulk_upsert(rows, column_types)
 
 
-def do_handle(event: Dict, _) -> Response:
+def do_handle(event: dict, _) -> Response:
     if "queryStringParameters" not in event:
         return BadRequest("Incorrect function call: non HTTP request")
 

examples/topic/writer_async_example.py

@@ -1,6 +1,5 @@
 import asyncio
 import datetime
-from typing import Dict, List
 
 import ydb
 from ydb import TopicWriterMessage
@@ -95,8 +94,8 @@ async def send_messages_and_wait_all_commit_with_results(
     await writer.flush()
 
 
-async def switch_messages_with_many_producers(writers: Dict[str, ydb.TopicWriterAsyncIO], messages: List[str]):
-    futures = []  # type:  List[asyncio.Future]
+async def switch_messages_with_many_producers(writers: dict[str, ydb.TopicWriterAsyncIO], messages: list[str]):
+    futures = []  # type:  list[asyncio.Future]
 
     for msg in messages:
         # select writer for the msg

examples/topic/writer_example.py

@@ -1,6 +1,5 @@
 import concurrent.futures
 import datetime
-from typing import Dict, List
 from concurrent.futures import Future, wait  # noqa: F401
 
 import ydb
@@ -123,7 +122,7 @@ def send_messages_and_wait_all_commit_with_flush(writer: ydb.TopicWriter):
 
 
 def send_messages_and_wait_all_commit_with_results(writer: ydb.TopicWriter):
-    futures = []  # type: List[concurrent.futures.Future]
+    futures = []  # type: list[concurrent.futures.Future]
     for i in range(10):
         future = writer.async_write_with_ack()
         futures.append(future)
@@ -134,8 +133,8 @@ def send_messages_and_wait_all_commit_with_results(writer: ydb.TopicWriter):
             raise future.exception()
 
 
-def switch_messages_with_many_producers(writers: Dict[str, ydb.TopicWriter], messages: List[str]):
-    futures = []  # type:  List[Future]
+def switch_messages_with_many_producers(writers: dict[str, ydb.TopicWriter], messages: list[str]):
+    futures = []  # type:  list[Future]
 
     for msg in messages:
         # select writer for the msg

generate_protoc.py

@@ -2,13 +2,12 @@
 import pathlib
 import shutil
 
-from typing import List
 from argparse import ArgumentParser
 
 from grpc_tools import command
 
 
-def files_filter(dir, items: List[str]) -> List[str]:
+def files_filter(dir, items: list[str]) -> list[str]:
     ignored_names = ['.git']
 
     ignore = []

tests/aio/query/test_query_session_pool.py

@@ -4,7 +4,6 @@
 import pytest
 import ydb
 
-from typing import Optional
 
 from ydb import QueryExplainResultFormat
 from ydb.aio.query.pool import QuerySessionPool
@@ -106,7 +105,7 @@ async def callee(session: QuerySession):
         ],
     )
     @pytest.mark.asyncio
-    async def test_retry_tx_normal(self, pool: QuerySessionPool, tx_mode: Optional[ydb.BaseQueryTxMode]):
+    async def test_retry_tx_normal(self, pool: QuerySessionPool, tx_mode: ydb.BaseQueryTxMode | None):
         retry_no = 0
 
         async def callee(tx: QueryTxContext):

tests/observability/test_observability_enable.py

@@ -11,7 +11,7 @@
 * ``get_trace_metadata`` is empty until a provider is enabled.
 """
 
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any
 
 import pytest
 
@@ -44,12 +44,12 @@
 class RecordingSpan:
     """Minimal Span implementation that records every call."""
 
-    def __init__(self, name: str, attributes: Optional[dict], kind: Optional[str], sink: List[Dict[str, Any]]):
+    def __init__(self, name: str, attributes: dict | None, kind: str | None, sink: list[dict[str, Any]]):
         self.name = name
-        self.attributes: Dict[str, Any] = dict(attributes or {})
+        self.attributes: dict[str, Any] = dict(attributes or {})
         self.kind = kind
         self.ended = False
-        self.errors: List[BaseException] = []
+        self.errors: list[BaseException] = []
         self._sink = sink
 
     def set_error(self, exception):
@@ -91,9 +91,9 @@ def __exit__(self_inner, exc_type, exc_val, exc_tb):
 class RecordingProvider:
     """Custom TracingProvider used across the tests."""
 
-    def __init__(self, metadata: Optional[List[Tuple[str, str]]] = None):
-        self.spans: List[RecordingSpan] = []
-        self.finished: List[Dict[str, Any]] = []
+    def __init__(self, metadata: list[tuple[str, str]] | None = None):
+        self.spans: list[RecordingSpan] = []
+        self.finished: list[dict[str, Any]] = []
         self._metadata = metadata or []
 
     def create_span(self, name, attributes=None, kind=None) -> Span:
@@ -423,7 +423,7 @@ class TestSetPeerAttributes:
     """Direct unit tests for ``set_peer_attributes``."""
 
     def _recording_span(self):
-        recorded: Dict[str, Any] = {}
+        recorded: dict[str, Any] = {}
 
         class _Span:
             def set_attribute(self, key, value):
@@ -460,7 +460,7 @@ class TestSpanFinishCallback:
     """``span_finish_callback`` wires stream completion into span lifecycle."""
 
     def test_finish_ends_span_on_success(self):
-        calls: List[str] = []
+        calls: list[str] = []
 
         class _Span:
             def set_error(self, exc):
@@ -473,7 +473,7 @@ def end(self):
         assert calls == ["end"]
 
     def test_finish_records_error_then_ends_span(self):
-        calls: List[str] = []
+        calls: list[str] = []
         exc = RuntimeError("stream broke")
 
         class _Span:
@@ -563,7 +563,7 @@ def test_set_attribute_forwards_to_underlying_otel_span(self):
 
         from ydb.opentelemetry.plugin import TracingSpan
 
-        recorded: Dict[str, Any] = {}
+        recorded: dict[str, Any] = {}
 
         class _FakeOtelSpan:
             def set_attribute(self, key, value):

tests/query/test_query_session_pool.py

@@ -5,7 +5,6 @@
 import time
 from concurrent import futures
 
-from typing import Optional
 
 from ydb import QueryExplainResultFormat
 from ydb.query.pool import QuerySessionPool
@@ -102,7 +101,7 @@ def callee(session: QuerySession):
             (ydb.QueryStaleReadOnly()),
         ],
     )
-    def test_retry_tx_normal(self, pool: QuerySessionPool, tx_mode: Optional[ydb.BaseQueryTxMode]):
+    def test_retry_tx_normal(self, pool: QuerySessionPool, tx_mode: ydb.BaseQueryTxMode | None):
         retry_no = 0
 
         def callee(tx: QueryTxContext):

tests/slo/src/core/metrics.py

@@ -8,7 +8,7 @@
 from contextlib import contextmanager
 from importlib.metadata import version
 from os import environ
-from typing import Any, Optional, Tuple
+from typing import Any
 
 OP_TYPE_READ, OP_TYPE_WRITE = "read", "write"
 OP_STATUS_SUCCESS, OP_STATUS_FAILURE = "success", "error"
@@ -19,7 +19,7 @@
 logger = logging.getLogger(__name__)
 
 
-def _normalize_labels(labels: Any) -> Tuple[Any, ...]:
+def _normalize_labels(labels: Any) -> tuple[Any, ...]:
     if labels is None:
         return tuple()
     if isinstance(labels, str):
@@ -44,7 +44,7 @@ def stop(
         labels,
         start_time: float,
         attempts: int = 1,
-        error: Optional[Exception] = None,
+        error: Exception | None = None,
     ) -> None:
         pass
 
@@ -95,7 +95,7 @@ def stop(
         labels,
         start_time: float,
         attempts: int = 1,
-        error: Optional[Exception] = None,
+        error: Exception | None = None,
     ) -> None:
         return None
 
@@ -228,7 +228,7 @@ def stop(
         labels,
         start_time: float,
         attempts: int = 1,
-        error: Optional[Exception] = None,
+        error: Exception | None = None,
     ) -> None:
         labels_t = _normalize_labels(labels)
         duration = time.time() - start_time
@@ -294,7 +294,7 @@ def inc_duplicated(self, n: int = 1) -> None:
         self._topic_duplicated.add(int(n), attributes={"ref": REF})
 
 
-def _resolve_metrics_endpoint(cli_endpoint: Optional[str]) -> str:
+def _resolve_metrics_endpoint(cli_endpoint: str | None) -> str:
     """
     Resolution order:
       1. OTEL_EXPORTER_OTLP_METRICS_ENDPOINT (used as-is)
@@ -315,7 +315,7 @@ def _resolve_metrics_endpoint(cli_endpoint: Optional[str]) -> str:
     return (cli_endpoint or "").strip()
 
 
-def create_metrics(otlp_endpoint: Optional[str]) -> BaseMetrics:
+def create_metrics(otlp_endpoint: str | None) -> BaseMetrics:
     """
     Build a metrics exporter.
 

tests/slo/src/root_runner.py

@@ -2,7 +2,6 @@
 import ydb
 import ydb.aio
 import logging
-from typing import Dict
 
 from core.metrics import WORKLOAD
 from runners.topic_runner import TopicRunner
@@ -20,7 +19,7 @@ def _is_async_workload(args) -> bool:
 
 class SLORunner:
     def __init__(self):
-        self.runners: Dict[str, type(BaseRunner)] = {}
+        self.runners: dict[str, type(BaseRunner)] = {}
 
     def register_runner(self, prefix: str, runner_cls: type(BaseRunner)):
         self.runners[prefix] = runner_cls

tests/slo/src/runners/base.py

@@ -1,14 +1,13 @@
 import logging
 from abc import ABC, abstractmethod
-from typing import Optional
 
 import ydb
 
 
 class BaseRunner(ABC):
     def __init__(self):
         self.logger = logging.getLogger(self.__class__.__module__)
-        self.driver: Optional[ydb.Driver] = None
+        self.driver: ydb.Driver | None = None
 
     @property
     @abstractmethod