22import re
33import tempfile
44from dataclasses import dataclass
5- from typing import Any , ClassVar , TypeAlias
5+ from typing import Annotated , Final , TypeAlias
66
77import sh # type: ignore[import-untyped]
88from models_library .docker import DockerGenericTag
99from pydantic import (
1010 BaseModel ,
1111 ByteSize ,
12- ConstrainedStr ,
12+ ConfigDict ,
1313 Field ,
1414 NonNegativeFloat ,
1515 NonNegativeInt ,
16- validator ,
16+ StringConstraints ,
17+ field_validator ,
1718)
1819from types_aiobotocore_ec2 .literals import InstanceStateNameType , InstanceTypeType
1920
@@ -33,26 +34,26 @@ def __gt__(self, other: "Resources") -> bool:
3334 return self .cpus > other .cpus or self .ram > other .ram
3435
3536 def __add__ (self , other : "Resources" ) -> "Resources" :
36- return Resources .construct (
37+ return Resources .model_construct (
3738 ** {
3839 key : a + b
3940 for (key , a ), b in zip (
40- self .dict ().items (), other .dict ().values (), strict = True
41+ self .model_dump ().items (), other .model_dump ().values (), strict = True
4142 )
4243 }
4344 )
4445
4546 def __sub__ (self , other : "Resources" ) -> "Resources" :
46- return Resources .construct (
47+ return Resources .model_construct (
4748 ** {
4849 key : a - b
4950 for (key , a ), b in zip (
50- self .dict ().items (), other .dict ().values (), strict = True
51+ self .model_dump ().items (), other .model_dump ().values (), strict = True
5152 )
5253 }
5354 )
5455
55- @validator ("cpus" , pre = True )
56+ @field_validator ("cpus" , mode = "before" )
5657 @classmethod
5758 def _floor_cpus_to_0 (cls , v : float ) -> float :
5859 return max (v , 0 )
@@ -67,19 +68,31 @@ class EC2InstanceType:
6768InstancePrivateDNSName : TypeAlias = str
6869
6970
70- class AWSTagKey (ConstrainedStr ):
71+ AWS_TAG_KEY_MIN_LENGTH : Final [int ] = 1
72+ AWS_TAG_KEY_MAX_LENGTH : Final [int ] = 128
73+ AWSTagKey : TypeAlias = Annotated [
7174 # see [https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html#tag-restrictions]
72- regex = re .compile (r"^(?!(_index|\.{1,2})$)[a-zA-Z0-9\+\-=\._:@]+$" )
73- min_length = 1
74- max_length = 128
75-
76-
77- class AWSTagValue (ConstrainedStr ):
75+ str ,
76+ StringConstraints (
77+ min_length = AWS_TAG_KEY_MIN_LENGTH ,
78+ max_length = AWS_TAG_KEY_MAX_LENGTH ,
79+ pattern = re .compile (r"^(?!(_index|\.{1,2})$)[a-zA-Z0-9\+\-=\._:@]+$" ),
80+ ),
81+ ]
82+
83+
84+ AWS_TAG_VALUE_MIN_LENGTH : Final [int ] = 0
85+ AWS_TAG_VALUE_MAX_LENGTH : Final [int ] = 256
86+ AWSTagValue : TypeAlias = Annotated [
7887 # see [https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html#tag-restrictions]
7988 # quotes []{} were added as it allows to json encode. it seems to be accepted as a value
80- regex = re .compile (r"^[a-zA-Z0-9\s\+\-=\.,_:/@\"\'\[\]\{\}]*$" )
81- min_length = 0
82- max_length = 256
89+ str ,
90+ StringConstraints (
91+ min_length = 0 ,
92+ max_length = 256 ,
93+ pattern = r"^[a-zA-Z0-9\s\+\-=\.,_:/@\"\'\[\]\{\}]*$" ,
94+ ),
95+ ]
8396
8497
8598EC2Tags : TypeAlias = dict [AWSTagKey , AWSTagValue ]
@@ -148,8 +161,23 @@ class EC2InstanceBootSpecific(BaseModel):
148161 default = 0 , description = "number of buffer EC2s to keep (defaults to 0)"
149162 )
150163
151- class Config :
152- schema_extra : ClassVar [dict [str , Any ]] = {
164+ @field_validator ("custom_boot_scripts" )
165+ @classmethod
166+ def validate_bash_calls (cls , v ):
167+ try :
168+ with tempfile .NamedTemporaryFile (mode = "wt" , delete = True ) as temp_file :
169+ temp_file .writelines (v )
170+ temp_file .flush ()
171+ # NOTE: this will not capture runtime errors, but at least some syntax errors such as invalid quotes
172+ sh .bash ("-n" , temp_file .name )
173+ except sh .ErrorReturnCode as exc :
174+ msg = f"Invalid bash call in custom_boot_scripts: { v } { exc .stderr }
175+ raise ValueError (msg ) from exc
176+
177+ return v
178+
179+ model_config = ConfigDict (
180+ json_schema_extra = {
153181 "examples" : [
154182 {
155183 # just AMI
@@ -205,18 +233,4 @@ class Config:
205233 },
206234 ]
207235 }
208-
209- @validator ("custom_boot_scripts" )
210- @classmethod
211- def validate_bash_calls (cls , v ):
212- try :
213- with tempfile .NamedTemporaryFile (mode = "wt" , delete = True ) as temp_file :
214- temp_file .writelines (v )
215- temp_file .flush ()
216- # NOTE: this will not capture runtime errors, but at least some syntax errors such as invalid quotes
217- sh .bash ("-n" , temp_file .name )
218- except sh .ErrorReturnCode as exc :
219- msg = f"Invalid bash call in custom_boot_scripts: { v } { exc .stderr }
220- raise ValueError (msg ) from exc
221-
222- return v
236+ )
0 commit comments