continue upgrading

Author: giancarloromeo

packages/aws-library/src/aws_library/ec2/__init__.py

@@ -1,6 +1,10 @@
 from ._client import SimcoreEC2API
 from ._errors import EC2AccessError, EC2NotConnectedError, EC2RuntimeError
 from ._models import (
+    AWS_TAG_KEY_MAX_LENGTH,
+    AWS_TAG_KEY_MIN_LENGTH,
+    AWS_TAG_VALUE_MAX_LENGTH,
+    AWS_TAG_VALUE_MIN_LENGTH,
     AWSTagKey,
     AWSTagValue,
     EC2InstanceBootSpecific,
@@ -14,6 +18,10 @@
 __all__: tuple[str, ...] = (
     "AWSTagKey",
     "AWSTagValue",
+    "AWS_TAG_KEY_MIN_LENGTH",
+    "AWS_TAG_KEY_MAX_LENGTH",
+    "AWS_TAG_VALUE_MIN_LENGTH",
+    "AWS_TAG_VALUE_MAX_LENGTH",
     "EC2AccessError",
     "EC2InstanceBootSpecific",
     "EC2InstanceConfig",

packages/aws-library/src/aws_library/ec2/_models.py

@@ -2,7 +2,7 @@
 import re
 import tempfile
 from dataclasses import dataclass
-from typing import Annotated, TypeAlias
+from typing import Annotated, Final, TypeAlias
 
 import sh  # type: ignore[import-untyped]
 from models_library.docker import DockerGenericTag
@@ -68,17 +68,21 @@ class EC2InstanceType:
 InstancePrivateDNSName: TypeAlias = str
 
 
+AWS_TAG_KEY_MIN_LENGTH: Final[int] = 1
+AWS_TAG_KEY_MAX_LENGTH: Final[int] = 128
 AWSTagKey: TypeAlias = Annotated[
     # see [https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html#tag-restrictions]
     str,
     StringConstraints(
-        min_length=1,
-        max_length=128,
+        min_length=AWS_TAG_KEY_MIN_LENGTH,
+        max_length=AWS_TAG_KEY_MAX_LENGTH,
         pattern=re.compile(r"^(?!(_index|\.{1,2})$)[a-zA-Z0-9\+\-=\._:@]+$"),
     ),
 ]
 
 
+AWS_TAG_VALUE_MIN_LENGTH: Final[int] = 0
+AWS_TAG_VALUE_MAX_LENGTH: Final[int] = 256
 AWSTagValue: TypeAlias = Annotated[
     # see [https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html#tag-restrictions]
     # quotes []{} were added as it allows to json encode. it seems to be accepted as a value

services/autoscaling/src/simcore_service_autoscaling/utils/buffer_machines_pool_core.py

@@ -1,11 +1,11 @@
 from collections.abc import Iterable
 from operator import itemgetter
 
-from aws_library.ec2 import AWSTagKey, AWSTagValue, EC2Tags
+from aws_library.ec2 import AWS_TAG_VALUE_MAX_LENGTH, AWSTagKey, AWSTagValue, EC2Tags
 from fastapi import FastAPI
 from models_library.docker import DockerGenericTag
 from models_library.utils.json_serialization import json_dumps
-from pydantic import parse_obj_as, parse_raw_as
+from pydantic import TypeAdapter
 
 from ..constants import (
     ACTIVATED_BUFFER_MACHINE_EC2_TAGS,
@@ -29,8 +29,10 @@ def get_deactivated_buffer_ec2_tags(
     base_ec2_tags = (
         auto_scaling_mode.get_ec2_tags(app) | DEACTIVATED_BUFFER_MACHINE_EC2_TAGS
     )
-    base_ec2_tags[AWSTagKey("Name")] = AWSTagValue(
-        f"{base_ec2_tags[AWSTagKey('Name')]}-buffer"
+    base_ec2_tags[TypeAdapter(AWSTagKey).validate_python("Name")] = TypeAdapter(
+        AWSTagValue
+    ).validate_python(
+        f"{base_ec2_tags[TypeAdapter(AWSTagKey).validate_python('Name')]}-buffer"
     )
     return base_ec2_tags
 
@@ -43,28 +45,36 @@ def dump_pre_pulled_images_as_tags(images: Iterable[DockerGenericTag]) -> EC2Tag
     # AWS Tag Values are limited to 256 characaters so we chunk the images
     # into smaller chunks
     jsonized_images = json_dumps(images)
-    assert AWSTagValue.max_length  # nosec
-    if len(jsonized_images) > AWSTagValue.max_length:
+    assert AWS_TAG_VALUE_MAX_LENGTH  # nosec
+    if len(jsonized_images) > AWS_TAG_VALUE_MAX_LENGTH:
         # let's chunk the string
-        chunk_size = AWSTagValue.max_length
+        chunk_size = AWS_TAG_VALUE_MAX_LENGTH
         chunks = [
             jsonized_images[i : i + chunk_size]
             for i in range(0, len(jsonized_images), chunk_size)
         ]
         return {
-            AWSTagKey(f"{PRE_PULLED_IMAGES_EC2_TAG_KEY}_({i})"): AWSTagValue(c)
+            TypeAdapter(AWSTagKey)
+            .validate_python(f"{PRE_PULLED_IMAGES_EC2_TAG_KEY}_({i})"): TypeAdapter(
+                AWSTagValue
+            )
+            .validate_python(c)
             for i, c in enumerate(chunks)
         }
     return {
-        PRE_PULLED_IMAGES_EC2_TAG_KEY: parse_obj_as(AWSTagValue, json_dumps(images))
+        PRE_PULLED_IMAGES_EC2_TAG_KEY: TypeAdapter(AWSTagValue).validate_python(
+            json_dumps(images)
+        )
     }
 
 
 def load_pre_pulled_images_from_tags(tags: EC2Tags) -> list[DockerGenericTag]:
     # AWS Tag values are limited to 256 characters so we chunk the images
     if PRE_PULLED_IMAGES_EC2_TAG_KEY in tags:
         # read directly
-        return parse_raw_as(list[DockerGenericTag], tags[PRE_PULLED_IMAGES_EC2_TAG_KEY])
+        return TypeAdapter(list[DockerGenericTag]).validate_json(
+            tags[PRE_PULLED_IMAGES_EC2_TAG_KEY]
+        )
 
     assembled_json = "".join(
         map(
@@ -80,5 +90,5 @@ def load_pre_pulled_images_from_tags(tags: EC2Tags) -> list[DockerGenericTag]:
         )
     )
     if assembled_json:
-        return parse_raw_as(list[DockerGenericTag], assembled_json)
+        return TypeAdapter(list[DockerGenericTag]).validate_json(assembled_json)
     return []