diff --git a/optuna_dashboard/_app.py b/optuna_dashboard/_app.py index 84642d9a..ac31e857 100644 --- a/optuna_dashboard/_app.py +++ b/optuna_dashboard/_app.py @@ -12,8 +12,6 @@ import typing from typing import Any from typing import Callable from typing import cast -from typing import Dict -from typing import List from typing import Optional from typing import TypeVar from typing import Union @@ -50,7 +48,7 @@ if typing.TYPE_CHECKING: except ImportError: FrozenStudy = None # type: ignore -BottleViewReturn = Union[str, bytes, Dict[str, Any], BaseResponse] +BottleViewReturn = Union[str, bytes, dict[str, Any], BaseResponse] BottleView = TypeVar("BottleView", bound=Callable[..., BottleViewReturn]) logger = logging.getLogger(__name__) @@ -63,8 +61,8 @@ cached_path_exists = functools.lru_cache(maxsize=10)(os.path.exists) # In-memory trials cache trials_cache_lock = threading.Lock() -trials_cache: Dict[int, List[FrozenTrial]] = {} -trials_last_fetched_at: Dict[int, datetime] = {} +trials_cache: dict[int, list[FrozenTrial]] = {} +trials_last_fetched_at: dict[int, datetime] = {} # RDB schema migration check rdb_schema_migrate_lock = threading.Lock() @@ -127,7 +125,7 @@ def update_schema_compatibility_flags(storage: BaseStorage) -> None: def json_api_view(view: BottleView) -> BottleView: @functools.wraps(view) - def decorated(*args: List[Any], **kwargs: Dict[str, Any]) -> BottleViewReturn: + def decorated(*args: list[Any], **kwargs: dict[str, Any]) -> BottleViewReturn: try: response.content_type = "application/json" response_body = view(*args, **kwargs) @@ -142,7 +140,7 @@ def json_api_view(view: BottleView) -> BottleView: return cast(BottleView, decorated) -def get_study_summaries(storage: BaseStorage) -> List[StudySummary]: +def get_study_summaries(storage: BaseStorage) -> list[StudySummary]: if version.parse(optuna_ver) >= version.Version("3.0.0rc0.dev"): frozen_studies = storage.get_all_studies() # type: ignore return [_frozen_study_to_study_summary(s) for s in frozen_studies] @@ -162,7 +160,7 @@ def get_study_summary(storage: BaseStorage, study_id: int) -> Optional[StudySumm def create_new_study( - storage: BaseStorage, study_name: str, directions: List[StudyDirection] + storage: BaseStorage, study_name: str, directions: list[StudyDirection] ) -> int: if version.parse(optuna_ver) >= version.Version("3.1.0.dev") and version.parse( optuna_ver @@ -174,7 +172,7 @@ def create_new_study( return study_id -def get_trials(storage: BaseStorage, study_id: int, ttl_seconds: int = 10) -> List[FrozenTrial]: +def get_trials(storage: BaseStorage, study_id: int, ttl_seconds: int = 10) -> list[FrozenTrial]: with trials_cache_lock: trials = trials_cache.get(study_id, None) last_fetched_at = trials_last_fetched_at.get(study_id, None) diff --git a/optuna_dashboard/_cached_extra_study_property.py b/optuna_dashboard/_cached_extra_study_property.py index 765e7833..105a1428 100644 --- a/optuna_dashboard/_cached_extra_study_property.py +++ b/optuna_dashboard/_cached_extra_study_property.py @@ -2,28 +2,24 @@ from __future__ import annotations import copy import threading -from typing import Dict -from typing import List from typing import Optional -from typing import Set -from typing import Tuple from optuna.distributions import BaseDistribution from optuna.trial import FrozenTrial from optuna.trial import TrialState -SearchSpaceSetT = Set[Tuple[str, BaseDistribution]] -SearchSpaceListT = List[Tuple[str, BaseDistribution]] +SearchSpaceSetT = set[tuple[str, BaseDistribution]] +SearchSpaceListT = list[tuple[str, BaseDistribution]] # In-memory cache cached_extra_study_property_cache_lock = threading.Lock() -cached_extra_study_property_cache: Dict[int, "_CachedExtraStudyProperty"] = {} +cached_extra_study_property_cache: dict[int, "_CachedExtraStudyProperty"] = {} def get_cached_extra_study_property( - study_id: int, trials: List[FrozenTrial] -) -> Tuple[SearchSpaceListT, SearchSpaceListT, List[Tuple[str, bool]], bool]: + study_id: int, trials: list[FrozenTrial] +) -> tuple[SearchSpaceListT, SearchSpaceListT, list[tuple[str, bool]], bool]: with cached_extra_study_property_cache_lock: cached_extra_study_property = cached_extra_study_property_cache.get(study_id, None) if cached_extra_study_property is None: @@ -45,7 +41,7 @@ class _CachedExtraStudyProperty: # union_user_attrs. self._intersection: Optional[SearchSpaceSetT] = None self._union: SearchSpaceSetT = set() - self._union_user_attrs: Dict[str, bool] = {} # attr_name: is_sortable (= is_number) + self._union_user_attrs: dict[str, bool] = {} # attr_name: is_sortable (= is_number) self.has_intermediate_values: bool = False @property @@ -63,12 +59,12 @@ class _CachedExtraStudyProperty: return union @property - def union_user_attrs(self) -> List[Tuple[str, bool]]: + def union_user_attrs(self) -> list[tuple[str, bool]]: union = [(name, is_sortable) for name, is_sortable in self._union_user_attrs.items()] sorted(union, key=lambda x: x[0]) return union - def update(self, trials: List[FrozenTrial]) -> None: + def update(self, trials: list[FrozenTrial]) -> None: next_cursor = self._cursor for trial in reversed(trials): if self._cursor > trial.number: diff --git a/optuna_dashboard/_importance.py b/optuna_dashboard/_importance.py index 94bbd51b..f6e0e4fa 100644 --- a/optuna_dashboard/_importance.py +++ b/optuna_dashboard/_importance.py @@ -1,9 +1,6 @@ from __future__ import annotations import threading -from typing import Dict -from typing import List -from typing import Tuple from typing import TYPE_CHECKING import warnings @@ -40,31 +37,31 @@ if TYPE_CHECKING: "ImportanceType", { "target_name": str, - "param_importances": List[ImportanceItemType], + "param_importances": list[ImportanceItemType], }, ) target_name = "Objective Value" param_importance_cache_lock = threading.Lock() # { "{study_id}:{objective_id}" : (n_completed_trials, importance) } -param_importance_cache: Dict[str, Tuple[int, ImportanceType]] = {} +param_importance_cache: dict[str, tuple[int, ImportanceType]] = {} class StudyWrapper(Study): def __init__( - self, storage: BaseStorage, study_id: int, cached_trials: List[FrozenTrial] + self, storage: BaseStorage, study_id: int, cached_trials: list[FrozenTrial] ) -> None: study_name = storage.get_study_name_from_id(study_id) super().__init__(study_name=study_name, storage=storage) self._cached_trials = cached_trials @property - def trials(self) -> List[FrozenTrial]: + def trials(self) -> list[FrozenTrial]: return self._cached_trials def get_param_importance_from_trials_cache( - storage: BaseStorage, study_id: int, objective_id: int, trials: List[FrozenTrial] + storage: BaseStorage, study_id: int, objective_id: int, trials: list[FrozenTrial] ) -> ImportanceType: completed_trials = [t for t in trials if t.state == TrialState.COMPLETE] n_completed_trials = len(completed_trials) @@ -97,7 +94,7 @@ def get_param_importance_from_trials_cache( def convert_to_importance_type( - importance: Dict[str, float], trials: List[FrozenTrial] + importance: dict[str, float], trials: list[FrozenTrial] ) -> ImportanceType: return { "target_name": target_name, @@ -112,7 +109,7 @@ def convert_to_importance_type( } -def get_distribution_name(param_name: str, trials: List[FrozenTrial]) -> str: +def get_distribution_name(param_name: str, trials: list[FrozenTrial]) -> str: for trial in trials: if param_name in trial.distributions: return trial.distributions[param_name].__class__.__name__ diff --git a/optuna_dashboard/_note.py b/optuna_dashboard/_note.py index 8b52e139..00710276 100644 --- a/optuna_dashboard/_note.py +++ b/optuna_dashboard/_note.py @@ -2,7 +2,6 @@ from __future__ import annotations import math from typing import Any -from typing import Dict from typing import TYPE_CHECKING from optuna.storages import BaseStorage @@ -24,20 +23,20 @@ NOTE_VER_KEY = "dashboard:note_ver" NOTE_STR_KEY_PREFIX = "dashboard:note_str:" -def get_note_from_system_attrs(system_attrs: Dict[str, Any]) -> NoteType: +def get_note_from_system_attrs(system_attrs: dict[str, Any]) -> NoteType: if NOTE_VER_KEY not in system_attrs: return { "version": 0, "body": "", } note_ver = int(system_attrs[NOTE_VER_KEY]) - note_attrs: Dict[str, str] = { + note_attrs: dict[str, str] = { key: value for key, value in system_attrs.items() if key.startswith(NOTE_STR_KEY_PREFIX) } return {"version": note_ver, "body": concat_body(note_attrs)} -def version_is_incremented(system_attrs: Dict[str, Any], req_note_ver: int) -> bool: +def version_is_incremented(system_attrs: dict[str, Any], req_note_ver: int) -> bool: db_note_ver = system_attrs.get(NOTE_VER_KEY, 0) return req_note_ver == db_note_ver + 1 @@ -50,7 +49,7 @@ def save_note(storage: BaseStorage, study_id: int, ver: int, body: str) -> None: storage.set_study_system_attr(study_id, k, v) # Clear previous messages - all_note_attrs: Dict[str, str] = { + all_note_attrs: dict[str, str] = { key: value for key, value in storage.get_study_system_attrs(study_id).items() if key.startswith(NOTE_STR_KEY_PREFIX) @@ -60,7 +59,7 @@ def save_note(storage: BaseStorage, study_id: int, ver: int, body: str) -> None: storage.set_study_system_attr(study_id, f"{NOTE_STR_KEY_PREFIX}{i}", "") -def split_body(note_str: str) -> Dict[str, str]: +def split_body(note_str: str) -> dict[str, str]: note_len = len(note_str) attrs = {} for i in range(math.ceil(note_len / SYSTEM_ATTR_MAX_LENGTH)): @@ -70,5 +69,5 @@ def split_body(note_str: str) -> Dict[str, str]: return attrs -def concat_body(note_attrs: Dict[str, str]) -> str: +def concat_body(note_attrs: dict[str, str]) -> str: return "".join(note_attrs[f"{NOTE_STR_KEY_PREFIX}{i}"] for i in range(len(note_attrs))) diff --git a/optuna_dashboard/_serializer.py b/optuna_dashboard/_serializer.py index 703848e5..9f42e83d 100644 --- a/optuna_dashboard/_serializer.py +++ b/optuna_dashboard/_serializer.py @@ -2,9 +2,6 @@ from __future__ import annotations import json from typing import Any -from typing import Dict -from typing import List -from typing import Tuple from typing import TYPE_CHECKING from typing import Union @@ -46,8 +43,8 @@ if TYPE_CHECKING: MAX_ATTR_LENGTH = 1024 -def serialize_attrs(attrs: Dict[str, Any]) -> List[Attribute]: - serialized: List[Attribute] = [] +def serialize_attrs(attrs: dict[str, Any]) -> list[Attribute]: + serialized: list[Attribute] = [] for k, v in attrs.items(): value: str if isinstance(v, bytes): @@ -59,7 +56,7 @@ def serialize_attrs(attrs: Dict[str, Any]) -> List[Attribute]: return serialized -def serialize_study_summary(summary: StudySummary) -> Dict[str, Any]: +def serialize_study_summary(summary: StudySummary) -> dict[str, Any]: serialized = { "study_id": summary._study_id, "study_name": summary.study_name, @@ -76,13 +73,13 @@ def serialize_study_summary(summary: StudySummary) -> Dict[str, Any]: def serialize_study_detail( summary: StudySummary, - trials: List[FrozenTrial], - intersection: List[Tuple[str, BaseDistribution]], - union: List[Tuple[str, BaseDistribution]], - union_user_attrs: List[Tuple[str, bool]], + trials: list[FrozenTrial], + intersection: list[tuple[str, BaseDistribution]], + union: list[tuple[str, BaseDistribution]], + union_user_attrs: list[tuple[str, bool]], has_intermediate_values: bool, -) -> Dict[str, Any]: - serialized: Dict[str, Any] = { +) -> dict[str, Any]: + serialized: dict[str, Any] = { "name": summary.study_name, "directions": [d.name.lower() for d in summary.directions], } @@ -98,7 +95,7 @@ def serialize_study_detail( return serialized -def serialize_frozen_trial(study_id: int, trial: FrozenTrial) -> Dict[str, Any]: +def serialize_frozen_trial(study_id: int, trial: FrozenTrial) -> dict[str, Any]: serialized = { "trial_id": trial._trial_id, "study_id": study_id, @@ -109,7 +106,7 @@ def serialize_frozen_trial(study_id: int, trial: FrozenTrial) -> Dict[str, Any]: "system_attrs": serialize_attrs(getattr(trial, "_system_attrs", {})), } - serialized_intermediate_values: List[IntermediateValue] = [] + serialized_intermediate_values: list[IntermediateValue] = [] for step, value in trial.intermediate_values.items(): serialized_value: Union[float, Literal["nan", "inf", "-inf"]] if np.isnan(value): @@ -127,7 +124,7 @@ def serialize_frozen_trial(study_id: int, trial: FrozenTrial) -> Dict[str, Any]: ) if trial.values is not None: - serialized_values: List[Union[float, Literal["inf", "-inf"]]] = [] + serialized_values: list[Union[float, Literal["inf", "-inf"]]] = [] for v in trial.values: assert not np.isnan(v), "Should not detect nan value" if np.isposinf(v): @@ -148,8 +145,8 @@ def serialize_frozen_trial(study_id: int, trial: FrozenTrial) -> Dict[str, Any]: def serialize_search_space( - search_space: List[Tuple[str, BaseDistribution]] -) -> List[Dict[str, Any]]: + search_space: list[tuple[str, BaseDistribution]] +) -> list[dict[str, Any]]: serialized = [] for param_name, distribution in search_space: serialized.append( diff --git a/optuna_dashboard/_sql_profiler.py b/optuna_dashboard/_sql_profiler.py index 3435dd22..40e34ca9 100644 --- a/optuna_dashboard/_sql_profiler.py +++ b/optuna_dashboard/_sql_profiler.py @@ -2,9 +2,6 @@ from __future__ import annotations import threading from time import perf_counter -from typing import Dict -from typing import List -from typing import Tuple from typing import TYPE_CHECKING from bottle import Bottle @@ -18,7 +15,7 @@ if TYPE_CHECKING: from sqlalchemy.engine.base import Engine sql_queries_lock = threading.Lock() -sql_queries: Dict[str, Tuple[int, List[float]]] = {} +sql_queries: dict[str, tuple[int, list[float]]] = {} sql_queries_template = SimpleTemplate( """ diff --git a/python_tests/test_cached_extra_study_property.py b/python_tests/test_cached_extra_study_property.py index bf21416b..16860586 100644 --- a/python_tests/test_cached_extra_study_property.py +++ b/python_tests/test_cached_extra_study_property.py @@ -1,6 +1,6 @@ +from __future__ import annotations + from typing import Any -from typing import Dict -from typing import List from unittest import TestCase import warnings @@ -19,7 +19,7 @@ class _CachedExtraStudyPropertySearchSpaceTestCase(TestCase): warnings.simplefilter("ignore", category=ExperimentalWarning) def test_same_distributions(self) -> None: - distributions: List[Dict[str, BaseDistribution]] = [ + distributions: list[dict[str, BaseDistribution]] = [ { "x0": FloatDistribution(low=0, high=10), "x1": FloatDistribution(low=0, high=10), @@ -50,7 +50,7 @@ class _CachedExtraStudyPropertySearchSpaceTestCase(TestCase): self.assertEqual(len(cached_extra_study_property.union), 2) def test_different_distributions(self) -> None: - distributions: List[Dict[str, BaseDistribution]] = [ + distributions: list[dict[str, BaseDistribution]] = [ { "x0": FloatDistribution(low=0, high=10), "x1": FloatDistribution(low=0, high=10), @@ -81,7 +81,7 @@ class _CachedExtraStudyPropertySearchSpaceTestCase(TestCase): self.assertEqual(len(cached_extra_study_property.union), 3) def test_dynamic_search_space(self) -> None: - distributions: List[Dict[str, BaseDistribution]] = [ + distributions: list[dict[str, BaseDistribution]] = [ { "x0": FloatDistribution(low=0, high=10), "x1": FloatDistribution(low=0, high=10), @@ -119,7 +119,7 @@ class _CachedExtraStudyPropertySearchSpaceTestCase(TestCase): self.assertEqual(len(cached_extra_study_property.union), 3) def test_contains_failed_trials(self) -> None: - distributions: Dict[str, BaseDistribution] = { + distributions: dict[str, BaseDistribution] = { "x0": FloatDistribution(low=0, high=10), "x1": FloatDistribution(low=0, high=10), } @@ -149,7 +149,7 @@ class _CachedExtraStudyPropertyIntermediateTestCase(TestCase): warnings.simplefilter("ignore", category=ExperimentalWarning) def test_no_intermediate_value(self) -> None: - intermediate_values: List[Dict] = [ + intermediate_values: list[dict] = [ {}, {}, ] @@ -168,7 +168,7 @@ class _CachedExtraStudyPropertyIntermediateTestCase(TestCase): self.assertFalse(cached_extra_study_property.has_intermediate_values) def test_some_trials_has_no_intermediate_value(self) -> None: - intermediate_values: List[Dict] = [ + intermediate_values: list[dict] = [ {0: 0.3, 1: 1.2}, {}, {0: 0.3, 1: 1.2}, @@ -188,7 +188,7 @@ class _CachedExtraStudyPropertyIntermediateTestCase(TestCase): self.assertTrue(cached_extra_study_property.has_intermediate_values) def test_all_trials_has_intermediate_value(self) -> None: - intermediate_values: List[Dict] = [{0: 0.3, 1: 1.2}, {0: 0.3, 1: 1.2}] + intermediate_values: list[dict] = [{0: 0.3, 1: 1.2}, {0: 0.3, 1: 1.2}] trials = [ create_trial( state=TrialState.COMPLETE, @@ -216,7 +216,7 @@ class _CachedExtraStudyPropertyUserAttrs(TestCase): warnings.simplefilter("ignore", category=ExperimentalWarning) def test_contains_failed_trials(self) -> None: - distributions: Dict[str, BaseDistribution] = { + distributions: dict[str, BaseDistribution] = { "x0": FloatDistribution(low=0, high=10), "x1": FloatDistribution(low=0, high=10), } @@ -253,7 +253,7 @@ class _CachedExtraStudyPropertyUserAttrs(TestCase): self.assertEqual(len(cached_extra_study_property.union_user_attrs), 3) def test_infer_sortable(self) -> None: - user_attrs_list: List[Dict[str, Any]] = [ + user_attrs_list: list[dict[str, Any]] = [ {"a": 1, "b": 1, "c": 1, "d": "a", "e": 1}, {"a": 2, "b": "a", "c": "a", "d": "a"}, {"a": 3, "b": None, "c": 3, "d": "a", "e": 3}, diff --git a/python_tests/wsgi_client.py b/python_tests/wsgi_client.py index 2a0a860a..36d48dfd 100644 --- a/python_tests/wsgi_client.py +++ b/python_tests/wsgi_client.py @@ -1,9 +1,6 @@ import io import typing -from typing import Dict -from typing import List from typing import Optional -from typing import Tuple from typing import Union from bottle import Bottle @@ -18,8 +15,8 @@ def create_wsgi_env( method: str, content_type: str, body: bytes, - queries: Dict[str, str], - headers: Dict[str, str], + queries: dict[str, str], + headers: dict[str, str], ) -> "WSGIEnvironment": # 'key1=value1&key2=value2' query_string = "&".join([f"{k}={v}" for k, v in queries.items()]) @@ -51,14 +48,14 @@ def send_request( path: str, method: str, body: Union[str, bytes] = b"", - queries: Optional[Dict[str, str]] = None, - headers: Optional[Dict[str, str]] = None, + queries: Optional[dict[str, str]] = None, + headers: Optional[dict[str, str]] = None, content_type: str = "text/plain; charset=utf-8", -) -> Tuple[int, List[Tuple[str, str]], bytes]: +) -> tuple[int, list[tuple[str, str]], bytes]: status: str = "" - response_headers: List[Tuple[str, str]] = [] + response_headers: list[tuple[str, str]] = [] - def start_response(status_: str, headers_: List[Tuple[str, str]]) -> None: + def start_response(status_: str, headers_: list[tuple[str, str]]) -> None: nonlocal status, response_headers status = status_ response_headers = headers_ diff --git a/visual_regression_test.py b/visual_regression_test.py index aa1050da..89e70f23 100644 --- a/visual_regression_test.py +++ b/visual_regression_test.py @@ -1,11 +1,10 @@ +from __future__ import annotations import argparse import asyncio import os import sys import threading import time -from typing import List -from typing import Tuple from wsgiref.simple_server import make_server import optuna @@ -89,7 +88,7 @@ def create_dummy_storage() -> optuna.storages.InMemoryStorage: directions=["minimize", "minimize"], ) - def objective_multi(trial: optuna.Trial) -> Tuple[float, float]: + def objective_multi(trial: optuna.Trial) -> tuple[float, float]: x = trial.suggest_float("x", 0, 5) y = trial.suggest_float("y", 0, 3) v0 = 4 * x**2 + 4 * y**2 @@ -103,7 +102,7 @@ def create_dummy_storage() -> optuna.storages.InMemoryStorage: study_name="multi-dynamic", storage=storage, directions=["minimize", "minimize"] ) - def objective_multi_dynamic(trial: optuna.Trial) -> Tuple[float, float]: + def objective_multi_dynamic(trial: optuna.Trial) -> tuple[float, float]: category = trial.suggest_categorical("category", ["foo", "bar"]) if category == "foo": x = trial.suggest_float("x1", 0, 5) @@ -173,8 +172,8 @@ async def contains_study_name(page: Page, study_name: str) -> bool: return False -async def take_screenshots(storage: optuna.storages.BaseStorage) -> List[str]: - validation_errors: List[str] = [] +async def take_screenshots(storage: optuna.storages.BaseStorage) -> list[str]: + validation_errors: list[str] = [] browser = await launch() page = await browser.newPage()