Upgrade type annotations syntax

We use the syntax from PEP 585 and PEP 604 and rely on 'from __future__
import annotations' to modernize the code base.

pgactivity/activities.py

@@ -1,7 +1,9 @@
+from __future__ import annotations
+
 import builtins
 import os
 import time
-from typing import Dict, List, Optional, Sequence, Tuple, TypeVar
+from typing import Sequence, TypeVar
 from warnings import catch_warnings, simplefilter
 
 import attr
@@ -21,7 +23,7 @@
 )
 
 
-def sys_get_proc(pid: int) -> Optional[SystemProcess]:
+def sys_get_proc(pid: int) -> SystemProcess | None:
     """Return a SystemProcess instance matching given pid or None if access with psutil
     is not possible.
     """
@@ -53,9 +55,9 @@ def sys_get_proc(pid: int) -> Optional[SystemProcess]:
 
 def ps_complete(
     pg_processes: Sequence[RunningProcess],
-    processes: Dict[int, SystemProcess],
+    processes: dict[int, SystemProcess],
     fs_blocksize: int,
-) -> Tuple[List[LocalRunningProcess], IOCounter, IOCounter]:
+) -> tuple[list[LocalRunningProcess], IOCounter, IOCounter]:
     """Transform the sequence of 'pg_processes' (RunningProcess) as LocalRunningProcess
     with local system information from the 'processes' map. Return LocalRunningProcess
     list, as well as read and write IO counters.
@@ -139,7 +141,7 @@ def ps_complete(
 T = TypeVar("T", RunningProcess, WaitingProcess, BlockingProcess, LocalRunningProcess)
 
 
-def sorted(processes: List[T], *, key: SortKey, reverse: bool = False) -> List[T]:
+def sorted(processes: list[T], *, key: SortKey, reverse: bool = False) -> list[T]:
     """Return processes sorted.
 
     >>> from ipaddress import IPv4Interface, ip_address
@@ -314,12 +316,12 @@ def update_max_iops(max_iops: int, read_count: float, write_count: float) -> int
     return max(int(read_count + write_count), max_iops)
 
 
-def get_load_average() -> Tuple[float, float, float]:
+def get_load_average() -> tuple[float, float, float]:
     """Get load average"""
     return os.getloadavg()
 
 
-def get_mem_swap() -> Tuple[MemoryInfo, SwapInfo]:
+def get_mem_swap() -> tuple[MemoryInfo, SwapInfo]:
     """Get memory and swap usage"""
     with catch_warnings():
         simplefilter("ignore", RuntimeWarning)
@@ -335,7 +337,7 @@ def get_mem_swap() -> Tuple[MemoryInfo, SwapInfo]:
     )
 
 
-def mem_swap_load() -> Tuple[MemoryInfo, SwapInfo, LoadAverage]:
+def mem_swap_load() -> tuple[MemoryInfo, SwapInfo, LoadAverage]:
     """Read memory, swap and load average from Data object."""
     memory, swap = get_mem_swap()
     load = LoadAverage(*get_load_average())

pgactivity/cli.py

@@ -1,11 +1,12 @@
+from __future__ import annotations
+
 import logging
 import os
 import socket
 import sys
 import time
 from argparse import ArgumentParser
 from io import StringIO
-from typing import Optional
 
 from blessed import Terminal
 
@@ -14,7 +15,7 @@
 from .pg import OperationalError
 
 
-def configure_logger(debug_file: Optional[str] = None) -> StringIO:
+def configure_logger(debug_file: str | None = None) -> StringIO:
     logger = logging.getLogger("pgactivity")
     logger.setLevel(logging.DEBUG)
 

pgactivity/compat.py

@@ -1,6 +1,8 @@
+from __future__ import annotations
+
 import operator
 from importlib.metadata import version
-from typing import Any, Dict
+from typing import Any
 
 import attr
 import attr.validators
@@ -11,7 +13,7 @@
 
 if ATTR_VERSION < (18, 1):
 
-    def fields_dict(cls: Any) -> Dict[str, Any]:
+    def fields_dict(cls: Any) -> dict[str, Any]:
         return {a.name: a for a in cls.__attrs_attrs__}
 
 else:

pgactivity/config.py

@@ -1,8 +1,10 @@
+from __future__ import annotations
+
 import configparser
 import enum
 import os
 from pathlib import Path
-from typing import IO, Any, Dict, List, Optional, Type, TypeVar
+from typing import IO, Any, Type, TypeVar
 
 import attr
 from attr import validators
@@ -70,22 +72,22 @@ class Flag(enum.Flag):
     PID = enum.auto()
 
     @classmethod
-    def names(cls) -> List[str]:
+    def names(cls) -> list[str]:
         rv = []
         for f in cls:
             assert f.name
             rv.append(f.name.lower())
         return rv
 
     @classmethod
-    def all(cls) -> "Flag":
+    def all(cls) -> Flag:
         value = cls(0)
         for f in cls:
             value |= f
         return value
 
     @classmethod
-    def from_config(cls, config: "Configuration") -> "Flag":
+    def from_config(cls, config: Configuration) -> Flag:
         value = cls(0)
         for f in cls:
             assert f.name is not None
@@ -102,7 +104,7 @@ def from_config(cls, config: "Configuration") -> "Flag":
     @classmethod
     def load(
         cls,
-        config: Optional["Configuration"],
+        config: Configuration | None,
         *,
         is_local: bool,
         noappname: bool,
@@ -117,7 +119,7 @@ def load(
         nowait: bool,
         nowrite: bool,
         **kwargs: Any,
-    ) -> "Flag":
+    ) -> Flag:
         """Build a Flag value from command line options."""
         if config:
             flag = cls.from_config(config)
@@ -163,13 +165,13 @@ def load(
 @attr.s(auto_attribs=True, frozen=True, slots=True)
 class UISection:
     hidden: bool = False
-    width: Optional[int] = attr.ib(default=None, validator=validators.optional(gt(0)))
+    width: int | None = attr.ib(default=None, validator=validators.optional(gt(0)))
 
     _T = TypeVar("_T", bound="UISection")
 
     @classmethod
     def from_config_section(cls: Type[_T], section: configparser.SectionProxy) -> _T:
-        values: Dict[str, Any] = {}
+        values: dict[str, Any] = {}
         known_options = {f.name: f for f in attr.fields(cls)}
         unknown_options = set(section) - set(known_options)
         if unknown_options:
@@ -192,7 +194,7 @@ def from_config_section(cls: Type[_T], section: configparser.SectionProxy) -> _T
 ETC = Path("/etc")
 
 
-class Configuration(Dict[str, UISection]):
+class Configuration(dict[str, UISection]):
     _T = TypeVar("_T", bound="Configuration")
 
     @classmethod
@@ -264,7 +266,7 @@ def lookup(
         *,
         user_config_home: Path = USER_CONFIG_HOME,
         etc: Path = ETC,
-    ) -> Optional[_T]:
+    ) -> _T | None:
         for base in (user_config_home, etc):
             fpath = base / "pg_activity.conf"
             if fpath.exists():

pgactivity/data.py

@@ -1,9 +1,10 @@
+from __future__ import annotations
+
 import getpass
 import logging
 import re
 from argparse import Namespace
 from functools import partial
-from typing import Dict, List, Optional
 
 import attr
 import psutil
@@ -70,24 +71,24 @@ class Data:
     server_encoding: bytes
     min_duration: float
     filters: Filters
-    dsn_parameters: Dict[str, str]
+    dsn_parameters: dict[str, str]
     failed_queries: FailedQueriesInfo
 
     @classmethod
     def pg_connect(
         cls,
         min_duration: float = 0.0,
         *,
-        host: Optional[str] = None,
+        host: str | None = None,
         port: int = 5432,
         user: str = "postgres",
-        password: Optional[str] = None,
+        password: str | None = None,
         database: str = "postgres",
         rds_mode: bool = False,
         dsn: str = "",
         hide_queries_in_logs: bool = False,
         filters: Filters = NO_FILTER,
-    ) -> "Data":
+    ) -> Data:
         """Create an instance by connecting to a PostgreSQL server."""
         pg_conn = pg.connect(
             dsn,
@@ -116,7 +117,7 @@ def pg_connect(
             dsn_parameters=pg.connection_parameters(pg_conn),
         )
 
-    def try_reconnect(self) -> Optional["Data"]:
+    def try_reconnect(self) -> Data | None:
         try:
             pg_conn = pg.connect(**self.dsn_parameters)
         except (pg.InterfaceError, pg.OperationalError):
@@ -189,7 +190,7 @@ def pg_terminate_backend(self, pid: int) -> bool:
         ret = pg.fetchone(self.pg_conn, query, {"pid": pid})
         return ret["is_stopped"]  # type: ignore[no-any-return]
 
-    def pg_get_temporary_file(self) -> Optional[TempFileInfo]:
+    def pg_get_temporary_file(self) -> TempFileInfo | None:
         """
         Count the number of temporary files and get their total size
         """
@@ -233,7 +234,7 @@ def pg_get_temporary_file(self) -> Optional[TempFileInfo]:
         finally:
             pg.execute(self.pg_conn, queries.get("reset_statement_timeout"))
 
-    def pg_get_wal_senders(self) -> Optional[int]:
+    def pg_get_wal_senders(self) -> int | None:
         """
         Count the number of wal senders
         """
@@ -244,7 +245,7 @@ def pg_get_wal_senders(self) -> Optional[int]:
         ret = pg.fetchone(self.pg_conn, query)
         return int(ret["wal_senders"])
 
-    def pg_get_wal_receivers(self) -> Optional[int]:
+    def pg_get_wal_receivers(self) -> int | None:
         """
         Count the number of wal receivers
         """
@@ -271,7 +272,7 @@ def pg_get_wal_receivers(self) -> Optional[int]:
 
         return int(ret["wal_receivers"])
 
-    def pg_get_replication_slots(self) -> Optional[int]:
+    def pg_get_replication_slots(self) -> int | None:
         """
         Count the number of replication slots
         """
@@ -290,7 +291,7 @@ def dbname_filter(self) -> sql.Composable:
 
     def pg_get_server_information(
         self,
-        prev_server_info: Optional[ServerInformation] = None,
+        prev_server_info: ServerInformation | None = None,
         using_rds: bool = False,
         skip_db_size: bool = False,
         skip_tempfile: bool = False,
@@ -338,17 +339,17 @@ def pg_get_server_information(
             )
             raise
 
-        temporary_file_info: Optional[TempFileInfo] = None
+        temporary_file_info: TempFileInfo | None = None
         if not skip_tempfile:
             temporary_file_info = self.pg_get_temporary_file()
         wal_senders = self.pg_get_wal_senders()
-        wal_receivers: Optional[int] = None
+        wal_receivers: int | None = None
         if not skip_walreceiver:
             wal_receivers = self.pg_get_wal_receivers()
         replication_slots = self.pg_get_replication_slots()
 
-        hr: Optional[Pct] = None
-        rr: Optional[Pct] = None
+        hr: Pct | None = None
+        rr: Pct | None = None
         tps, ips, ups, dps, rps = 0, 0, 0, 0, 0
         size_ev = 0.0
         if prev_server_info is not None:
@@ -390,7 +391,7 @@ def pg_get_server_information(
             **ret,
         )
 
-    def pg_get_activities(self, duration_mode: int = 1) -> List[RunningProcess]:
+    def pg_get_activities(self, duration_mode: int = 1) -> list[RunningProcess]:
         """
         Get activity from pg_stat_activity view.
         """
@@ -425,7 +426,7 @@ def pg_get_activities(self, duration_mode: int = 1) -> List[RunningProcess]:
             text_as_bytes=True,
         )
 
-    def pg_get_waiting(self, duration_mode: int = 1) -> List[WaitingProcess]:
+    def pg_get_waiting(self, duration_mode: int = 1) -> list[WaitingProcess]:
         """
         Get waiting queries.
         """
@@ -452,7 +453,7 @@ def pg_get_waiting(self, duration_mode: int = 1) -> List[WaitingProcess]:
             text_as_bytes=True,
         )
 
-    def pg_get_blocking(self, duration_mode: int = 1) -> List[BlockingProcess]:
+    def pg_get_blocking(self, duration_mode: int = 1) -> list[BlockingProcess]:
         """
         Get blocking queries
         """

pgactivity/handlers.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from __future__ import annotations
 
 from blessed.keyboard import Keystroke
 
@@ -8,7 +8,7 @@
 
 
 def refresh_time(
-    key: Optional[str], value: float, minimum: float = 0.5, maximum: float = 5
+    key: str | None, value: float, minimum: float = 0.5, maximum: float = 5
 ) -> float:
     """Return an updated refresh time interval from input key respecting bounds.
 
@@ -66,7 +66,7 @@ def wrap_query(key: Keystroke, wrap: bool) -> bool:
     return wrap
 
 
-def query_mode(key: Keystroke) -> Optional[QueryMode]:
+def query_mode(key: Keystroke) -> QueryMode | None:
     """Return the query mode matching input key or None.
 
     >>> import curses
@@ -86,9 +86,7 @@ def query_mode(key: Keystroke) -> Optional[QueryMode]:
     return keys.QUERYMODE_FROM_KEYS.get(key)
 
 
-def sort_key_for(
-    key: Keystroke, query_mode: QueryMode, flag: Flag
-) -> Optional[SortKey]:
+def sort_key_for(key: Keystroke, query_mode: QueryMode, flag: Flag) -> SortKey | None:
     """Return the sort key matching input key or None.
 
     >>> from blessed.keyboard import Keystroke as k