Ad ruff rules UP(pyupgrade)

astrapy/api_commander.py

@@ -21,11 +21,6 @@
     Any,
     Dict,
     Iterable,
-    List,
-    Optional,
-    Tuple,
-    Type,
-    Union,
     cast,
 )
 
@@ -75,8 +70,8 @@ def __init__(
         self,
         api_endpoint: str,
         path: str,
-        headers: Dict[str, Union[str, None]] = {},
-        callers: List[Tuple[Optional[str], Optional[str]]] = [],
+        headers: dict[str, str | None] = {},
+        callers: list[tuple[str | None, str | None]] = [],
         redacted_header_names: Iterable[str] = DEFAULT_REDACTED_HEADER_NAMES,
         dev_ops_api: bool = False,
     ) -> None:
@@ -88,15 +83,14 @@ def __init__(
         self.redacted_header_names = set(redacted_header_names)
         self.dev_ops_api = dev_ops_api
 
-        self._faulty_response_exc_class: Union[
-            Type[DevOpsAPIFaultyResponseException], Type[DataAPIFaultyResponseException]
-        ]
-        self._response_exc_class: Union[
-            Type[DevOpsAPIResponseException], Type[DataAPIResponseException]
-        ]
-        self._http_exc_class: Union[
-            Type[DataAPIHttpException], Type[DevOpsAPIHttpException]
-        ]
+        self._faulty_response_exc_class: (
+            type[DevOpsAPIFaultyResponseException]
+            | type[DataAPIFaultyResponseException]
+        )
+        self._response_exc_class: (
+            type[DevOpsAPIResponseException] | type[DataAPIResponseException]
+        )
+        self._http_exc_class: type[DataAPIHttpException] | type[DevOpsAPIHttpException]
         if self.dev_ops_api:
             self._faulty_response_exc_class = DevOpsAPIFaultyResponseException
             self._response_exc_class = DevOpsAPIResponseException
@@ -109,10 +103,10 @@ def __init__(
         full_user_agent_string = compose_full_user_agent(
             [user_agent_ragstack] + self.callers + [user_agent_astrapy]
         )
-        self.caller_header: Dict[str, str] = (
+        self.caller_header: dict[str, str] = (
             {"User-Agent": full_user_agent_string} if full_user_agent_string else {}
         )
-        self.full_headers: Dict[str, str] = {
+        self.full_headers: dict[str, str] = {
             **{k: v for k, v in self.headers.items() if v is not None},
             **self.caller_header,
             **{"Content-Type": "application/json"},
@@ -143,20 +137,20 @@ async def __aenter__(self) -> APICommander:
 
     async def __aexit__(
         self,
-        exc_type: Optional[Type[BaseException]] = None,
-        exc_value: Optional[BaseException] = None,
-        traceback: Optional[TracebackType] = None,
+        exc_type: type[BaseException] | None = None,
+        exc_value: BaseException | None = None,
+        traceback: TracebackType | None = None,
     ) -> None:
         await self.async_client.aclose()
 
     def _copy(
         self,
-        api_endpoint: Optional[str] = None,
-        path: Optional[str] = None,
-        headers: Optional[Dict[str, Union[str, None]]] = None,
-        callers: Optional[List[Tuple[Optional[str], Optional[str]]]] = None,
-        redacted_header_names: Optional[List[str]] = None,
-        dev_ops_api: Optional[bool] = None,
+        api_endpoint: str | None = None,
+        path: str | None = None,
+        headers: dict[str, str | None] | None = None,
+        callers: list[tuple[str | None, str | None]] | None = None,
+        redacted_header_names: list[str] | None = None,
+        dev_ops_api: bool | None = None,
     ) -> APICommander:
         # some care in allowing e.g. {} to override (but not None):
         return APICommander(
@@ -178,10 +172,10 @@ def _raw_response_to_json(
         self,
         raw_response: httpx.Response,
         raise_api_errors: bool,
-        payload: Optional[Dict[str, Any]],
-    ) -> Dict[str, Any]:
+        payload: dict[str, Any] | None,
+    ) -> dict[str, Any]:
         # try to process the httpx raw response into a JSON or throw a failure
-        raw_response_json: Dict[str, Any]
+        raw_response_json: dict[str, Any]
         try:
             raw_response_json = cast(
                 Dict[str, Any],
@@ -212,15 +206,15 @@ def _raw_response_to_json(
         response_json = restore_from_api(raw_response_json)
         return response_json
 
-    def _compose_request_url(self, additional_path: Optional[str]) -> str:
+    def _compose_request_url(self, additional_path: str | None) -> str:
         if additional_path:
             return "/".join([self.full_path.rstrip("/"), additional_path.lstrip("/")])
         else:
             return self.full_path
 
     def _encode_payload(
-        self, normalized_payload: Optional[Dict[str, Any]]
-    ) -> Optional[bytes]:
+        self, normalized_payload: dict[str, Any] | None
+    ) -> bytes | None:
         if normalized_payload is not None:
             return json.dumps(
                 normalized_payload,
@@ -234,8 +228,8 @@ def raw_request(
         self,
         *,
         http_method: str = HttpMethod.POST,
-        payload: Optional[Dict[str, Any]] = None,
-        additional_path: Optional[str] = None,
+        payload: dict[str, Any] | None = None,
+        additional_path: str | None = None,
         raise_api_errors: bool = True,
         timeout_info: TimeoutInfoWideType = None,
     ) -> httpx.Response:
@@ -276,8 +270,8 @@ async def async_raw_request(
         self,
         *,
         http_method: str = HttpMethod.POST,
-        payload: Optional[Dict[str, Any]] = None,
-        additional_path: Optional[str] = None,
+        payload: dict[str, Any] | None = None,
+        additional_path: str | None = None,
         raise_api_errors: bool = True,
         timeout_info: TimeoutInfoWideType = None,
     ) -> httpx.Response:
@@ -318,11 +312,11 @@ def request(
         self,
         *,
         http_method: str = HttpMethod.POST,
-        payload: Optional[Dict[str, Any]] = None,
-        additional_path: Optional[str] = None,
+        payload: dict[str, Any] | None = None,
+        additional_path: str | None = None,
         raise_api_errors: bool = True,
         timeout_info: TimeoutInfoWideType = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         raw_response = self.raw_request(
             http_method=http_method,
             payload=payload,
@@ -338,11 +332,11 @@ async def async_request(
         self,
         *,
         http_method: str = HttpMethod.POST,
-        payload: Optional[Dict[str, Any]] = None,
-        additional_path: Optional[str] = None,
+        payload: dict[str, Any] | None = None,
+        additional_path: str | None = None,
         raise_api_errors: bool = True,
         timeout_info: TimeoutInfoWideType = None,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         raw_response = await self.async_raw_request(
             http_method=http_method,
             payload=payload,

astrapy/api_options.py

@@ -15,7 +15,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Optional, TypeVar
+from typing import TypeVar
 
 from astrapy.authentication import (
     EmbeddingAPIKeyHeaderProvider,
@@ -41,9 +41,9 @@ class BaseAPIOptions:
             much sense.
     """
 
-    max_time_ms: Optional[int] = None
+    max_time_ms: int | None = None
 
-    def with_default(self: AO, default: Optional[BaseAPIOptions]) -> AO:
+    def with_default(self: AO, default: BaseAPIOptions | None) -> AO:
         """
         Return a new instance created by completing this instance with a default
         API options object.
@@ -70,7 +70,7 @@ def with_default(self: AO, default: Optional[BaseAPIOptions]) -> AO:
         else:
             return self
 
-    def with_override(self: AO, override: Optional[BaseAPIOptions]) -> AO:
+    def with_override(self: AO, override: BaseAPIOptions | None) -> AO:
         """
         Return a new instance created by overriding the members of this instance
         with those taken from a supplied "override" API options object.

astrapy/authentication.py

@@ -16,7 +16,7 @@
 
 import base64
 from abc import ABC, abstractmethod
-from typing import Any, Dict, Optional, Union
+from typing import Any
 
 from astrapy.defaults import (
     EMBEDDING_HEADER_API_KEY,
@@ -28,15 +28,15 @@
 )
 
 
-def coerce_token_provider(token: Optional[Union[str, TokenProvider]]) -> TokenProvider:
+def coerce_token_provider(token: str | TokenProvider | None) -> TokenProvider:
     if isinstance(token, TokenProvider):
         return token
     else:
         return StaticTokenProvider(token)
 
 
 def coerce_embedding_headers_provider(
-    embedding_api_key: Optional[Union[str, EmbeddingHeadersProvider]],
+    embedding_api_key: str | EmbeddingHeadersProvider | None,
 ) -> EmbeddingHeadersProvider:
     if isinstance(embedding_api_key, EmbeddingHeadersProvider):
         return embedding_api_key
@@ -121,7 +121,7 @@ def __bool__(self) -> bool:
         return self.get_token() is not None
 
     @abstractmethod
-    def get_token(self) -> Union[str, None]:
+    def get_token(self) -> str | None:
         """
         Produce a string for direct use as token in a subsequent API request,
         or None for no token.
@@ -146,7 +146,7 @@ class StaticTokenProvider(TokenProvider):
         ... )
     """
 
-    def __init__(self, token: Union[str, None]) -> None:
+    def __init__(self, token: str | None) -> None:
         self.token = token
 
     def __repr__(self) -> str:
@@ -155,7 +155,7 @@ def __repr__(self) -> str:
         else:
             return self.token
 
-    def get_token(self) -> Union[str, None]:
+    def get_token(self) -> str | None:
         return self.token
 
 
@@ -230,7 +230,7 @@ def __bool__(self) -> bool:
         return self.get_headers() != {}
 
     @abstractmethod
-    def get_headers(self) -> Dict[str, str]:
+    def get_headers(self) -> dict[str, str]:
         """
         Produce a dictionary for use as (part of) the headers in HTTP requests
         to the Data API.
@@ -277,7 +277,7 @@ class EmbeddingAPIKeyHeaderProvider(EmbeddingHeadersProvider):
         ... )
     """
 
-    def __init__(self, embedding_api_key: Optional[str]) -> None:
+    def __init__(self, embedding_api_key: str | None) -> None:
         self.embedding_api_key = embedding_api_key
 
     def __repr__(self) -> str:
@@ -286,7 +286,7 @@ def __repr__(self) -> str:
         else:
             return f'{self.__class__.__name__}("{redact_secret(self.embedding_api_key, 8)}")'
 
-    def get_headers(self) -> Dict[str, str]:
+    def get_headers(self) -> dict[str, str]:
         if self.embedding_api_key is not None:
             return {EMBEDDING_HEADER_API_KEY: self.embedding_api_key}
         else:
@@ -347,7 +347,7 @@ def __repr__(self) -> str:
             f'embedding_secret_id="{redact_secret(self.embedding_secret_id, 6)}")'
         )
 
-    def get_headers(self) -> Dict[str, str]:
+    def get_headers(self) -> dict[str, str]:
         return {
             EMBEDDING_HEADER_AWS_ACCESS_ID: self.embedding_access_id,
             EMBEDDING_HEADER_AWS_SECRET_ID: self.embedding_secret_id,