Skip to content

Commit

Permalink
Merge pull request optuna#5840 from boringbyte/fix/_redis.py
Browse files Browse the repository at this point in the history
Simplify type annotations for `optuna/storages/journal/_redis.py`
  • Loading branch information
not522 authored Dec 9, 2024
2 parents a92b168 + 0a5d2a1 commit ae849ba
Showing 1 changed file with 7 additions and 8 deletions.
15 changes: 7 additions & 8 deletions optuna/storages/journal/_redis.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
from __future__ import annotations

import json
import time
from typing import Any
from typing import Dict
from typing import List
from typing import Optional

from optuna._deprecated import deprecated_class
from optuna._experimental import experimental_class
Expand Down Expand Up @@ -44,16 +43,16 @@ def __init__(self, url: str, use_cluster: bool = False, prefix: str = "") -> Non
self._use_cluster = use_cluster
self._prefix = prefix

def __getstate__(self) -> Dict[Any, Any]:
def __getstate__(self) -> dict[Any, Any]:
state = self.__dict__.copy()
del state["_redis"]
return state

def __setstate__(self, state: Dict[Any, Any]) -> None:
def __setstate__(self, state: dict[Any, Any]) -> None:
self.__dict__.update(state)
self._redis = redis.Redis.from_url(self._url)

def read_logs(self, log_number_from: int) -> List[Dict[str, Any]]:
def read_logs(self, log_number_from: int) -> list[dict[str, Any]]:
max_log_number_bytes = self._redis.get(f"{self._prefix}:log_number")
if max_log_number_bytes is None:
return []
Expand All @@ -75,7 +74,7 @@ def read_logs(self, log_number_from: int) -> List[Dict[str, Any]]:
raise err
return logs

def append_logs(self, logs: List[Dict[str, Any]]) -> None:
def append_logs(self, logs: list[dict[str, Any]]) -> None:
self._redis.setnx(f"{self._prefix}:log_number", -1)
for log in logs:
if not self._use_cluster:
Expand All @@ -93,7 +92,7 @@ def append_logs(self, logs: List[Dict[str, Any]]) -> None:
def save_snapshot(self, snapshot: bytes) -> None:
self._redis.set(f"{self._prefix}:snapshot", snapshot)

def load_snapshot(self) -> Optional[bytes]:
def load_snapshot(self) -> bytes | None:
snapshot_bytes = self._redis.get(f"{self._prefix}:snapshot")
return snapshot_bytes

Expand Down

0 comments on commit ae849ba

Please sign in to comment.