Merge branch 'master' into prep/5.2.0

Author: bblommers

.github/workflows/release.yml

@@ -73,7 +73,7 @@ jobs:
         git tag ${{ env.VERSION }}
         git push origin ${{ env.VERSION }}
     - name: Create GitHub release
-      uses: softprops/action-gh-release@v2
+      uses: softprops/action-gh-release@v3
       with:
         name: ${{ env.VERSION }}
         tag_name: ${{ env.VERSION }}

.github/workflows/tests_sdk_ruby.yml

@@ -13,7 +13,7 @@ jobs:
     steps:
       - uses: actions/checkout@v6
       - name: Set up Ruby ${{ matrix.ruby-version }}
-        uses: ruby/setup-ruby@3ff19f5e2baf30647122352b96108b1fbe250c64
+        uses: ruby/setup-ruby@c4e5b1316158f92e3d49443a9d58b31d25ac0f8f
         with:
           ruby-version: ${{ matrix.ruby-version }}
       - name: Set up Python

moto/acm/models.py

@@ -4,7 +4,7 @@
 import ipaddress
 import re
 from collections.abc import Iterable
-from typing import Any, Optional
+from typing import Any
 
 import cryptography.hazmat.primitives.asymmetric.rsa
 import cryptography.x509
@@ -64,12 +64,12 @@ def datetime_to_epoch(date: datetime.datetime) -> float:
     return float(calendar.timegm(aware_dt.timetuple()))
 
 
-class TagHolder(dict[str, Optional[str]]):
+class TagHolder(dict[str, str | None]):
     MAX_TAG_COUNT = 50
     MAX_KEY_LENGTH = 128
     MAX_VALUE_LENGTH = 256
 
-    def _validate_kv(self, key: str, value: Optional[str], index: int) -> None:
+    def _validate_kv(self, key: str, value: str | None, index: int) -> None:
         if len(key) > self.MAX_KEY_LENGTH:
             raise AWSValidationException(
                 f"Value '{key}' at 'tags.{index}.member.key' failed to satisfy constraint: Member must have length less than or equal to {self.MAX_KEY_LENGTH}"
@@ -127,13 +127,13 @@ def __init__(
         account_id: str,
         certificate: bytes,
         private_key: bytes,
-        chain: Optional[bytes] = None,
+        chain: bytes | None = None,
         region: str = "us-east-1",
-        arn: Optional[str] = None,
+        arn: str | None = None,
         cert_type: str = "IMPORTED",
         cert_status: str = "ISSUED",
-        cert_authority_arn: Optional[str] = None,
-        cert_options: Optional[dict[str, Any]] = None,
+        cert_authority_arn: str | None = None,
+        cert_options: dict[str, Any] | None = None,
     ):
         self.created_at = utcnow()
         self.cert = certificate
@@ -175,8 +175,8 @@ def generate_cert(
         domain_name: str,
         account_id: str,
         region: str,
-        sans: Optional[list[str]] = None,
-        cert_authority_arn: Optional[str] = None,
+        sans: list[str] | None = None,
+        cert_authority_arn: str | None = None,
     ) -> "CertBundle":
         unique_sans: set[str] = set(sans) if sans else set()
 
@@ -477,7 +477,7 @@ def set_certificate_in_use_by(self, arn: str, load_balancer_name: str) -> None:
         cert_bundle = self._certificates[arn]
         cert_bundle.in_use_by.append(load_balancer_name)
 
-    def _get_arn_from_idempotency_token(self, token: str) -> Optional[str]:
+    def _get_arn_from_idempotency_token(self, token: str) -> str | None:
         """
         If token doesnt exist, return None, later it will be
         set with an expiry and arn.
@@ -510,8 +510,8 @@ def import_certificate(
         self,
         certificate: bytes,
         private_key: bytes,
-        chain: Optional[bytes],
-        arn: Optional[str],
+        chain: bytes | None,
+        arn: str | None,
         tags: list[dict[str, str]],
     ) -> str:
         if arn is not None:
@@ -594,8 +594,8 @@ def request_certificate(
         idempotency_token: str,
         subject_alt_names: list[str],
         tags: list[dict[str, str]],
-        cert_authority_arn: Optional[str] = None,
-        cert_options: Optional[dict[str, Any]] = None,
+        cert_authority_arn: str | None = None,
+        cert_options: dict[str, Any] | None = None,
     ) -> str:
         """
         The parameter DomainValidationOptions has not yet been implemented

moto/acm/responses.py

@@ -1,13 +1,12 @@
 import base64
 import json
-from typing import Union
 
 from moto.core.responses import BaseResponse
 
 from .exceptions import AWSValidationException
 from .models import AWSCertificateManagerBackend, acm_backends
 
-GENERIC_RESPONSE_TYPE = Union[str, tuple[str, dict[str, int]]]
+GENERIC_RESPONSE_TYPE = str | tuple[str, dict[str, int]]
 
 
 class AWSCertificateManagerResponse(BaseResponse):

moto/acmpca/models.py

@@ -3,7 +3,7 @@
 import base64
 import contextlib
 import datetime
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
@@ -35,7 +35,7 @@ def __init__(
         certificate_authority_configuration: dict[str, Any],
         certificate_authority_type: str,
         revocation_configuration: dict[str, Any],
-        security_standard: Optional[str],
+        security_standard: str | None,
     ):
         self.id = mock_random.uuid4()
         self.arn = f"arn:{get_partition(region)}:acm-pca:{region}:{account_id}:certificate-authority/{self.id}"
@@ -48,11 +48,11 @@ def __init__(
         }
         self.set_revocation_configuration(revocation_configuration)
         self.created_at = unix_time()
-        self.updated_at: Optional[float] = None
+        self.updated_at: float | None = None
         self.status = "PENDING_CERTIFICATE"
         self.usage_mode = "SHORT_LIVED_CERTIFICATE"
         self.security_standard = security_standard or "FIPS_140_2_LEVEL_3_OR_HIGHER"
-        self.policy: Optional[str] = None
+        self.policy: str | None = None
         self.revoked_certificates: dict[str, dict[str, Any]] = {}
 
         self.password = str(mock_random.uuid4()).encode("utf-8")
@@ -65,7 +65,7 @@ def __init__(
         )
 
         self.certificate_bytes: bytes = b""
-        self.certificate_chain: Optional[bytes] = None
+        self.certificate_chain: bytes | None = None
         self.issued_certificates: dict[str, bytes] = {}
         self.issued_certificates_certificate_chains: dict[str, bytes] = {}
 
@@ -103,7 +103,7 @@ def key(self) -> rsa.RSAPrivateKey:
         return cast(rsa.RSAPrivateKey, private_key)
 
     @property
-    def certificate(self) -> Optional[x509.Certificate]:
+    def certificate(self) -> x509.Certificate | None:
         if self.certificate_bytes:
             return x509.load_pem_x509_certificate(self.certificate_bytes)
         return None
@@ -153,7 +153,7 @@ def csr(self) -> bytes:
         )
         return csr.public_bytes(serialization.Encoding.PEM)
 
-    def issue_certificate(self, csr_bytes: bytes, template_arn: Optional[str]) -> str:
+    def issue_certificate(self, csr_bytes: bytes, template_arn: str | None) -> str:
         csr = x509.load_pem_x509_csr(base64.b64decode(csr_bytes))
         extensions = self._x509_extensions(csr, template_arn)
         new_cert = self.generate_cert(
@@ -177,7 +177,7 @@ def issue_certificate(self, csr_bytes: bytes, template_arn: Optional[str]) -> st
         return cert_arn
 
     def _x509_extensions(
-        self, csr: x509.CertificateSigningRequest, template_arn: Optional[str]
+        self, csr: x509.CertificateSigningRequest, template_arn: str | None
     ) -> list[tuple[x509.ExtensionType, bool]]:
         """
         Uses a PCA certificate template ARN to return a list of X.509 extensions.
@@ -283,7 +283,7 @@ def get_certificate(self, certificate_arn: str) -> tuple[bytes, bytes]:
         return certificate, certificate_chain
 
     def set_revocation_configuration(
-        self, revocation_configuration: Optional[dict[str, Any]]
+        self, revocation_configuration: dict[str, Any] | None
     ) -> None:
         if revocation_configuration is not None:
             self.revocation_configuration = revocation_configuration
@@ -300,7 +300,7 @@ def set_revocation_configuration(
                         raise InvalidS3ObjectAclInCrlConfiguration(acl)
 
     @property
-    def not_valid_after(self) -> Optional[float]:
+    def not_valid_after(self) -> float | None:
         if self.certificate is None:
             return None
         try:
@@ -309,7 +309,7 @@ def not_valid_after(self) -> Optional[float]:
             return unix_time(self.certificate.not_valid_after)
 
     @property
-    def not_valid_before(self) -> Optional[float]:
+    def not_valid_before(self) -> float | None:
         if self.certificate is None:
             return None
         try:
@@ -318,7 +318,7 @@ def not_valid_before(self) -> Optional[float]:
             return unix_time(self.certificate.not_valid_before)
 
     def import_certificate_authority_certificate(
-        self, certificate: bytes, certificate_chain: Optional[bytes]
+        self, certificate: bytes, certificate_chain: bytes | None
     ) -> None:
         try:
             x509.load_pem_x509_certificate(certificate)
@@ -376,7 +376,7 @@ def create_certificate_authority(
         certificate_authority_configuration: dict[str, Any],
         revocation_configuration: dict[str, Any],
         certificate_authority_type: str,
-        security_standard: Optional[str],
+        security_standard: str | None,
         tags: list[dict[str, str]],
     ) -> str:
         """
@@ -404,7 +404,7 @@ def describe_certificate_authority(
 
     def get_certificate_authority_certificate(
         self, certificate_authority_arn: str
-    ) -> tuple[bytes, Optional[bytes]]:
+    ) -> tuple[bytes, bytes | None]:
         ca = self.describe_certificate_authority(certificate_authority_arn)
         if ca.status != "ACTIVE":
             raise InvalidStateException(certificate_authority_arn)
@@ -439,7 +439,7 @@ def delete_certificate_authority(self, certificate_authority_arn: str) -> None:
         ca.status = "DELETED"
 
     def issue_certificate(
-        self, certificate_authority_arn: str, csr: bytes, template_arn: Optional[str]
+        self, certificate_authority_arn: str, csr: bytes, template_arn: str | None
     ) -> str:
         """
         The following parameters are not yet implemented: ApiPassthrough, SigningAlgorithm, Validity, ValidityNotBefore, IdempotencyToken
@@ -471,7 +471,7 @@ def import_certificate_authority_certificate(
         self,
         certificate_authority_arn: str,
         certificate: bytes,
-        certificate_chain: Optional[bytes],
+        certificate_chain: bytes | None,
     ) -> None:
         ca = self.describe_certificate_authority(certificate_authority_arn)
         ca.import_certificate_authority_certificate(certificate, certificate_chain)
@@ -532,7 +532,7 @@ def delete_policy(self, resource_arn: str) -> None:
     @paginate(pagination_model=PAGINATION_MODEL)
     def list_certificate_authorities(
         self,
-        resource_owner: Optional[str] = None,
+        resource_owner: str | None = None,
     ) -> list[CertificateAuthority]:
         """
         Lists the private certificate authorities that you created by using the CreateCertificateAuthority action.

moto/amp/models.py

@@ -1,7 +1,7 @@
 """PrometheusServiceBackend class with methods for supported APIs."""
 
 from collections.abc import Callable
-from typing import Any, Optional
+from typing import Any
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -64,7 +64,7 @@ def __init__(
         self.created_at = unix_time()
         self.tag_fn = tag_fn
         self.rule_group_namespaces: dict[str, RuleGroupNamespace] = {}
-        self.logging_config: Optional[dict[str, Any]] = None
+        self.logging_config: dict[str, Any] | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return {

moto/apigateway/integration_parsers/__init__.py

@@ -1,5 +1,4 @@
 import abc
-from typing import Union
 
 from requests.models import PreparedRequest
 
@@ -10,5 +9,5 @@ class IntegrationParser:
     @abc.abstractmethod
     def invoke(
         self, request: PreparedRequest, integration: Integration
-    ) -> tuple[int, Union[str, bytes]]:
+    ) -> tuple[int, str | bytes]:
         pass

moto/apigateway/integration_parsers/aws_parser.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import requests
 
 from ..models import Integration
@@ -9,7 +7,7 @@
 class TypeAwsParser(IntegrationParser):
     def invoke(
         self, request: requests.PreparedRequest, integration: Integration
-    ) -> tuple[int, Union[str, bytes]]:
+    ) -> tuple[int, str | bytes]:
         # integration.uri = arn:aws:apigateway:{region}:{subdomain.service|service}:path|action/{service_api}
         # example value = 'arn:aws:apigateway:us-west-2:dynamodb:action/PutItem'
         try:

moto/apigateway/integration_parsers/http_parser.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import requests
 
 from ..models import Integration
@@ -13,7 +11,7 @@ class TypeHttpParser(IntegrationParser):
 
     def invoke(
         self, request: requests.PreparedRequest, integration: Integration
-    ) -> tuple[int, Union[str, bytes]]:
+    ) -> tuple[int, str | bytes]:
         uri = integration.uri
         requests_func = getattr(requests, integration.http_method.lower())  # type: ignore[union-attr]
         response = requests_func(uri)

moto/apigateway/integration_parsers/unknown_parser.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import requests
 
 from ..models import Integration
@@ -13,6 +11,6 @@ class TypeUnknownParser(IntegrationParser):
 
     def invoke(
         self, request: requests.PreparedRequest, integration: Integration
-    ) -> tuple[int, Union[str, bytes]]:
+    ) -> tuple[int, str | bytes]:
         _type = integration.integration_type
         raise NotImplementedError(f"The {_type} type has not been implemented")