Merge branch 'develop' into add-cfn-schem-fetcher

Author: vicheey

requirements/base.txt

@@ -1,7 +1,8 @@
 boto3>=1.34.0,<2.0.0
-jsonschema<5,>=3.2  # TODO: evaluate risk of removing jsonschema 3.x support
+jsonschema<5,>=4.23
 typing_extensions>=4.4 # 3.8 doesn't have Required, TypeGuard and ParamSpec
 
 # resource validation & schema generation
-# 1.10.15 and 1.10.17 included breaking change from pydantic, more info: https://github.com/aws/serverless-application-model/issues/3617
-pydantic>=1.8,<3,!=1.10.15,!=1.10.17
+# we should remove support for Python 3.8 soon (and 3.9), but supporting it for now
+pydantic>=2.10.6; python_version>="3.8"
+pydantic~=2.12.5; python_version>="3.9"

requirements/dev.txt

@@ -23,12 +23,9 @@ black==24.3.0
 ruamel.yaml==0.17.21  # It can parse yaml while perserving comments
 
 # type check
-mypy~=1.3.0
+mypy~=1.10.1
 
 # types
 boto3-stubs[appconfig,serverlessrepo]>=1.34.0,<2.0.0
 types-PyYAML~=6.0
 types-jsonschema~=3.2
-
-# CloudFormation CLI tools
-cloudformation-cli>=0.2.39,<0.3.0

samtranslator/model/__init__.py

@@ -824,15 +824,14 @@ def resolve_resource_type(self, resource_dict: Dict[str, Any]) -> Any:
 
 
 class ResourceResolver:
-    def __init__(self, resources: Dict[str, Any]) -> None:
+    def __init__(self, resources: Dict[str, Dict[str, Any]]) -> None:
         """
         Instantiate the resolver
         :param dict resources: Map of resource
         """
 
         if not isinstance(resources, dict):
             raise TypeError("'Resources' is either null or not a valid dictionary.")
-
         self.resources = resources
 
     def get_all_resources(self) -> Dict[str, Any]:
@@ -850,7 +849,6 @@ def get_resource_by_logical_id(self, _input: str) -> Optional[Dict[str, Any]]:
         """
         if not isinstance(_input, str):
             raise TypeError(f"Invalid logical ID '{_input}'. Expected a string.")
-
         return self.resources.get(_input, None)
 
 

samtranslator/model/apigateway.py

@@ -522,7 +522,7 @@ def _get_identity_header(self) -> Optional[str]:
         if not self.identity or not self.identity.get("Header"):
             return "Authorization"
 
-        return self.identity.get("Header")
+        return self.identity.get("Header")  # type: ignore[no-any-return]
 
     def _get_reauthorize_every(self) -> Optional[PassThrough]:
         if not self.identity:

samtranslator/model/sam_resources.py

@@ -2,7 +2,9 @@
 
 import copy
 import re
+import sys
 from contextlib import suppress
+from types import ModuleType
 from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union, cast
 
 import samtranslator.model.eventsources
@@ -34,11 +36,19 @@
     SyncConfigType,
     UserPoolConfigType,
 )
-from samtranslator.internal.schema_source import (
-    aws_serverless_capacity_provider,
-    aws_serverless_function,
-    aws_serverless_graphqlapi,
-)
+
+# Pydantic 1 doesn't support Python 3.14 so these imports will fail until we migrate to v2
+try:
+    from samtranslator.internal.schema_source import (
+        aws_serverless_capacity_provider,
+        aws_serverless_function,
+        aws_serverless_graphqlapi,
+    )
+except RuntimeError:  # Pydantic fails when initializing the model classes with a RuntimeError in 3.14
+    aws_serverless_capacity_provider = cast(ModuleType, None)
+    aws_serverless_function = cast(ModuleType, None)
+    aws_serverless_graphqlapi = cast(ModuleType, None)
+
 from samtranslator.internal.schema_source.common import PermissionsType, SamIntrinsicable
 from samtranslator.internal.types import GetManagedPolicyMap
 from samtranslator.internal.utils.utils import passthrough_value, remove_none_values
@@ -143,6 +153,14 @@
 _CONDITION_CHAR_LIMIT = 255
 
 
+# Utility function to throw an error when using functionality that doesn't work in Python 3.14 (need migration to Pydantic v2)
+def check_python_314_compatibility(module: Optional[ModuleType], functionality: str) -> None:
+    if sys.version_info >= (3, 14) and module is None:
+        raise RuntimeError(
+            f"{functionality} functionalities are temporarily not supported when running SAM in Python 3.14"
+        )
+
+
 class SamFunction(SamResourceMacro):
     """SAM function macro."""
 
@@ -788,6 +806,7 @@ def _transform_capacity_provider_config(self) -> Dict[str, Any]:
 
         # Validate CapacityProviderConfig using Pydantic model directly for comprehensive error collection
         try:
+            check_python_314_compatibility(aws_serverless_function, "Capacity Provider")
             validated_model = aws_serverless_function.CapacityProviderConfig.parse_obj(self.CapacityProviderConfig)
         except Exception as e:
             raise InvalidResourceException(self.logical_id, f"Invalid CapacityProviderConfig: {e!s}") from e
@@ -1516,7 +1535,6 @@ class SamCapacityProvider(SamResourceMacro):
     """
 
     resource_type = "AWS::Serverless::CapacityProvider"
-    resource_property_schema = aws_serverless_capacity_provider.Properties
     property_types = {
         "CapacityProviderName": Property(False, one_of(IS_STR, IS_DICT)),
         "VpcConfig": Property(True, IS_DICT),
@@ -1549,7 +1567,11 @@ def to_cloudformation(self, **kwargs: Any) -> List[Resource]:
         """
         Transform the SAM CapacityProvider resource to CloudFormation
         """
-        self.validate_before_transform(schema_class=self.resource_property_schema, collect_all_errors=True)
+        check_python_314_compatibility(aws_serverless_capacity_provider, "Capacity Provider")
+        self.validate_before_transform(
+            schema_class=aws_serverless_capacity_provider.Properties,
+            collect_all_errors=True,
+        )
 
         # Use enhanced validation method with comprehensive error collection
         model = self.validate_properties_and_return_model(
@@ -2657,6 +2679,7 @@ def __init__(
 
     @cw_timer
     def to_cloudformation(self, **kwargs: Any) -> List[Resource]:
+        check_python_314_compatibility(aws_serverless_graphqlapi, "GraphQLApi")
         model = self.validate_properties_and_return_model(aws_serverless_graphqlapi.Properties)
 
         appsync_api, cloudwatch_role, auth_connectors = self._construct_appsync_api_resources(model)

samtranslator/model/stepfunctions/generators.py

@@ -396,7 +396,7 @@ def _get_paths_to_intrinsics(self, _input, path=None):  # type: ignore[no-untype
         else:
             return dynamic_value_paths
 
-        for key, value in sorted(iterator, key=lambda item: item[0]):  # type: ignore[no-any-return]
+        for key, value in sorted(iterator, key=lambda item: item[0]):
             if is_intrinsic(value) or is_dynamic_reference(value):
                 dynamic_value_paths.append([*path, key])
             elif isinstance(value, (dict, list)):