Pyupgrade stubs

.vscode/settings.json

@@ -1,6 +1,9 @@
 {
     "[python]": {
         "editor.formatOnSave": true,
-        "editor.defaultFormatter": "charliermarsh.ruff"
+        "editor.defaultFormatter": "charliermarsh.ruff",
+        "editor.codeActionsOnSave": {
+            "source.organizeImports": "explicit"
+        }
     }
 }

awscrt-stubs/__init__.pyi

@@ -1,3 +1,7 @@
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 from awscrt import auth as auth
 from awscrt import crypto as crypto
 from awscrt import http as http

awscrt-stubs/_test.pyi

@@ -1,3 +1,7 @@
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 from awscrt import NativeResource as NativeResource
 from awscrt.io import ClientBootstrap as ClientBootstrap
 from awscrt.io import DefaultHostResolver as DefaultHostResolver

awscrt-stubs/auth.pyi

@@ -1,7 +1,11 @@
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 from concurrent.futures import Future
 from datetime import datetime
 from enum import IntEnum
-from typing import Any, Callable, List, Optional, Sequence, Tuple, Type, TypeVar
+from typing import Any, Callable, Sequence, TypeVar
 
 from awscrt import NativeResource as NativeResource
 from awscrt.http import HttpProxyOptions
@@ -16,8 +20,8 @@ class AwsCredentials(NativeResource):
         self,
         access_key_id: str,
         secret_access_key: str,
-        session_token: Optional[str] = ...,
-        expiration: Optional[datetime] = ...,
+        session_token: str | None = ...,
+        expiration: datetime | None = ...,
     ) -> None: ...
     @property
     def access_key_id(self) -> str: ...
@@ -26,62 +30,60 @@ class AwsCredentials(NativeResource):
     @property
     def session_token(self) -> str: ...
     @property
-    def expiration(self) -> Optional[datetime]: ...
+    def expiration(self) -> datetime | None: ...
     def __deepcopy__(self: _R, memo: Any) -> _R: ...
 
 class AwsCredentialsProviderBase(NativeResource): ...
 
 class AwsCredentialsProvider(AwsCredentialsProviderBase):
     def __init__(self, binding: Any) -> None: ...
     @classmethod
-    def new_default_chain(
-        cls: Type[_R], client_bootstrap: Optional[ClientBootstrap] = ...
-    ) -> _R: ...
+    def new_default_chain(cls: type[_R], client_bootstrap: ClientBootstrap | None = ...) -> _R: ...
     @classmethod
     def new_static(
-        cls: Type[_R],
+        cls: type[_R],
         access_key_id: str,
         secret_access_key: str,
-        session_token: Optional[str] = ...,
+        session_token: str | None = ...,
     ) -> _R: ...
     @classmethod
     def new_profile(
-        cls: Type[_R],
-        client_bootstrap: Optional[ClientBootstrap] = ...,
-        profile_name: Optional[str] = ...,
-        config_filepath: Optional[str] = ...,
-        credentials_filepath: Optional[str] = ...,
+        cls: type[_R],
+        client_bootstrap: ClientBootstrap | None = ...,
+        profile_name: str | None = ...,
+        config_filepath: str | None = ...,
+        credentials_filepath: str | None = ...,
     ) -> _R: ...
     @classmethod
-    def new_process(cls: Type[_R], profile_to_use: Optional[str] = ...) -> _R: ...
+    def new_process(cls: type[_R], profile_to_use: str | None = ...) -> _R: ...
     @classmethod
-    def new_environment(cls: Type[_R]) -> _R: ...
+    def new_environment(cls: type[_R]) -> _R: ...
     @classmethod
-    def new_chain(cls: Type[_R], providers: List[AwsCredentialsProvider]) -> _R: ...
+    def new_chain(cls: type[_R], providers: list[AwsCredentialsProvider]) -> _R: ...
     @classmethod
-    def new_delegate(cls: Type[_R], get_credentials: Callable[[], AwsCredentials]) -> _R: ...
+    def new_delegate(cls: type[_R], get_credentials: Callable[[], AwsCredentials]) -> _R: ...
     @classmethod
     def new_cognito(
-        cls: Type[_R],
+        cls: type[_R],
         *,
         endpoint: str,
         identity: str,
         tls_ctx: ClientTlsContext,
-        logins: Optional[Sequence[Tuple[str, str]]] = ...,
-        custom_role_arn: Optional[str] = ...,
-        client_bootstrap: Optional[ClientBootstrap] = ...,
-        http_proxy_options: Optional[HttpProxyOptions] = ...,
+        logins: Sequence[tuple[str, str]] | None = ...,
+        custom_role_arn: str | None = ...,
+        client_bootstrap: ClientBootstrap | None = ...,
+        http_proxy_options: HttpProxyOptions | None = ...,
     ) -> _R: ...
     @classmethod
     def new_x509(
-        cls: Type[_R],
+        cls: type[_R],
         *,
         endpoint: str,
         thing_name: str,
         role_alias: str,
         tls_ctx: ClientTlsContext,
-        client_bootstrap: Optional[ClientBootstrap] = ...,
-        http_proxy_options: Optional[HttpProxyOptions] = ...,
+        client_bootstrap: ClientBootstrap | None = ...,
+        http_proxy_options: HttpProxyOptions | None = ...,
     ) -> _R: ...
     def get_credentials(self) -> Future[AwsCredentials]: ...
 
@@ -109,16 +111,16 @@ class AwsSigningConfig(NativeResource):
         self,
         algorithm: AwsSigningAlgorithm = ...,
         signature_type: AwsSignatureType = ...,
-        credentials_provider: Optional[AwsCredentialsProvider] = ...,
+        credentials_provider: AwsCredentialsProvider | None = ...,
         region: str = ...,
         service: str = ...,
-        date: Optional[datetime] = ...,
-        should_sign_header: Optional[Callable[[str], bool]] = ...,
+        date: datetime | None = ...,
+        should_sign_header: Callable[[str], bool] | None = ...,
         use_double_uri_encode: bool = ...,
         should_normalize_uri_path: bool = ...,
-        signed_body_value: Optional[str] = ...,
+        signed_body_value: str | None = ...,
         signed_body_header_type: AwsSignedBodyHeaderType = ...,
-        expiration_in_seconds: Optional[int] = ...,
+        expiration_in_seconds: int | None = ...,
         omit_session_token: bool = ...,
     ) -> None: ...
     def replace(self: _R, **kwargs: Any) -> _R: ...
@@ -135,17 +137,17 @@ class AwsSigningConfig(NativeResource):
     @property
     def date(self) -> datetime: ...
     @property
-    def should_sign_header(self) -> Optional[Callable[[str], bool]]: ...
+    def should_sign_header(self) -> Callable[[str], bool] | None: ...
     @property
     def use_double_uri_encode(self) -> bool: ...
     @property
     def should_normalize_uri_path(self) -> bool: ...
     @property
-    def signed_body_value(self) -> Optional[str]: ...
+    def signed_body_value(self) -> str | None: ...
     @property
     def signed_body_header_type(self) -> AwsSignedBodyHeaderType: ...
     @property
-    def expiration_in_seconds(self) -> Optional[int]: ...
+    def expiration_in_seconds(self) -> int | None: ...
     @property
     def omit_session_token(self) -> bool: ...
 

awscrt-stubs/crypto.pyi

@@ -1,5 +1,9 @@
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 from enum import IntEnum
-from typing import Any, Union
+from typing import Any
 
 from awscrt import NativeResource
 
@@ -35,36 +39,36 @@ class RSA(NativeResource):
     def __init__(self, binding: Any) -> None: ...
     @staticmethod
     def new_private_key_from_pem_data(
-        pem_data: Union[str, bytes, bytearray, memoryview],
-    ) -> "RSA": ...
+        pem_data: str | bytes | bytearray | memoryview,
+    ) -> RSA: ...
     @staticmethod
     def new_public_key_from_pem_data(
-        pem_data: Union[str, bytes, bytearray, memoryview],
-    ) -> "RSA": ...
+        pem_data: str | bytes | bytearray | memoryview,
+    ) -> RSA: ...
     @staticmethod
     def new_private_key_from_der_data(
-        der_data: Union[str, bytes, bytearray, memoryview],
-    ) -> "RSA": ...
+        der_data: str | bytes | bytearray | memoryview,
+    ) -> RSA: ...
     @staticmethod
-    def new_public_key_from_der_data(der_data: Union[bytes, bytearray, memoryview]) -> "RSA": ...
+    def new_public_key_from_der_data(der_data: bytes | bytearray | memoryview) -> RSA: ...
     def encrypt(
         self,
         encryption_algorithm: RSAEncryptionAlgorithm,
-        plaintext: Union[bytes, bytearray, memoryview],
+        plaintext: bytes | bytearray | memoryview,
     ) -> bytes: ...
     def decrypt(
         self,
         encryption_algorithm: RSAEncryptionAlgorithm,
-        ciphertext: Union[bytes, bytearray, memoryview],
+        ciphertext: bytes | bytearray | memoryview,
     ) -> bytes: ...
     def sign(
         self,
         signature_algorithm: RSASignatureAlgorithm,
-        digest: Union[bytes, bytearray, memoryview],
+        digest: bytes | bytearray | memoryview,
     ) -> bytes: ...
     def verify(
         self,
         signature_algorithm: RSASignatureAlgorithm,
-        digest: Union[bytes, bytearray, memoryview],
-        signature: Union[bytes, bytearray, memoryview],
+        digest: bytes | bytearray | memoryview,
+        signature: bytes | bytearray | memoryview,
     ) -> bool: ...

awscrt-stubs/eventstream/__init__.pyi

@@ -1,4 +1,7 @@
-from collections.abc import ByteString
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 from enum import IntEnum
 from typing import Any
 from uuid import UUID
@@ -29,7 +32,7 @@ class Header:
     @classmethod
     def from_int64(cls, name: str, value: int) -> Header: ...
     @classmethod
-    def from_byte_buf(cls, name: str, value: ByteString) -> Header: ...
+    def from_byte_buf(cls, name: str, value: bytes | bytearray) -> Header: ...
     @classmethod
     def from_string(cls, name: str, value: str) -> Header: ...
     @classmethod
@@ -47,7 +50,7 @@ class Header:
     def value_as_int16(self) -> int: ...
     def value_as_int32(self) -> int: ...
     def value_as_int64(self) -> int: ...
-    def value_as_byte_buf(self) -> ByteString: ...
+    def value_as_byte_buf(self) -> bytes | bytearray: ...
     def value_as_string(self) -> str: ...
     def value_as_timestamp(self) -> int: ...
     def value_as_uuid(self) -> UUID: ...

awscrt-stubs/eventstream/rpc.pyi

@@ -1,8 +1,12 @@
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 import abc
 from abc import ABC, abstractmethod
 from concurrent.futures import Future
 from enum import IntEnum
-from typing import Any, Callable, Optional, Sequence
+from typing import Any, Callable, Sequence
 
 from awscrt import NativeResource
 from awscrt.eventstream import Header
@@ -30,12 +34,12 @@ class ClientConnectionHandler(ABC, metaclass=abc.ABCMeta):
     @abstractmethod
     def on_connection_setup(
         self,
-        connection: Optional[HttpClientConnection],
-        error: Optional[BaseException],
+        connection: HttpClientConnection | None,
+        error: BaseException | None,
         **kwargs: Any,
     ) -> None: ...
     @abstractmethod
-    def on_connection_shutdown(self, reason: Optional[BaseException], **kwargs: Any) -> None: ...
+    def on_connection_shutdown(self, reason: BaseException | None, **kwargs: Any) -> None: ...
     @abstractmethod
     def on_protocol_message(
         self,
@@ -59,47 +63,47 @@ class ClientConnection(NativeResource):
         handler: ClientConnectionHandler,
         host_name: str,
         port: int,
-        bootstrap: Optional[ClientBootstrap] = ...,
-        socket_options: Optional[SocketOptions] = ...,
-        tls_connection_options: Optional[TlsConnectionOptions] = ...,
-    ) -> Future[Optional[BaseException]]: ...
-    def close(self) -> Future[Optional[BaseException]]: ...
+        bootstrap: ClientBootstrap | None = ...,
+        socket_options: SocketOptions | None = ...,
+        tls_connection_options: TlsConnectionOptions | None = ...,
+    ) -> Future[BaseException | None]: ...
+    def close(self) -> Future[BaseException | None]: ...
     def is_open(self) -> bool: ...
     def send_protocol_message(
         self,
         *,
-        headers: Optional[Sequence[Header]] = ...,
-        payload: Optional[bytes] = ...,
+        headers: Sequence[Header] | None = ...,
+        payload: bytes | None = ...,
         message_type: MessageType,
-        flags: Optional[int] = ...,
-        on_flush: Optional[Callable[..., Any]] = ...,
-    ) -> Future[Optional[BaseException]]: ...
+        flags: int | None = ...,
+        on_flush: Callable[..., Any] | None = ...,
+    ) -> Future[BaseException | None]: ...
     def new_stream(self, handler: ClientContinuationHandler) -> ClientContinuation: ...
 
 class ClientContinuation(NativeResource):
     def __init__(self, handler: ClientConnectionHandler, connection: ClientConnection) -> None:
         self.connection: ClientConnection
-        closed_future: Future[None]
+        self.closed_future: Future[None]
 
     def activate(
         self,
         *,
         operation: str,
-        headers: Optional[Sequence[Header]] = ...,
-        payload: Optional[bytes] = ...,
+        headers: Sequence[Header] | None = ...,
+        payload: bytes | None = ...,
         message_type: MessageType,
-        flags: Optional[int] = ...,
-        on_flush: Optional[Callable[..., Any]] = ...,
-    ) -> Future[Optional[BaseException]]: ...
+        flags: int | None = ...,
+        on_flush: Callable[..., Any] | None = ...,
+    ) -> Future[BaseException | None]: ...
     def send_message(
         self,
         *,
-        headers: Optional[Sequence[Header]] = ...,
-        payload: Optional[bytes] = ...,
+        headers: Sequence[Header] | None = ...,
+        payload: bytes | None = ...,
         message_type: MessageType,
-        flags: Optional[int] = ...,
-        on_flush: Optional[Callable[..., Any]] = ...,
-    ) -> Future[Optional[BaseException]]: ...
+        flags: int | None = ...,
+        on_flush: Callable[..., Any] | None = ...,
+    ) -> Future[BaseException | None]: ...
     def is_closed(self) -> bool: ...
 
 class ClientContinuationHandler(ABC, metaclass=abc.ABCMeta):

awscrt-stubs/exceptions.pyi

@@ -1,3 +1,7 @@
+"""
+Copyright 2024 Vlad Emelianov
+"""
+
 def from_code(code: int) -> AwsCrtError: ...
 
 class AwsCrtError(Exception):