migration

Author: sanderegg

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

@@ -29,7 +29,7 @@
     Task,
     TaskState,
 )
-from pydantic import ByteSize, ValidationError, parse_obj_as
+from pydantic import ByteSize, TypeAdapter, ValidationError
 from servicelib.docker_utils import to_datetime
 from servicelib.logging_utils import log_context
 from servicelib.utils import logged_gather
@@ -59,25 +59,27 @@
 _PENDING_DOCKER_TASK_MESSAGE: Final[str] = "pending task scheduling"
 _INSUFFICIENT_RESOURCES_DOCKER_TASK_ERR: Final[str] = "insufficient resources on"
 _NOT_SATISFIED_SCHEDULING_CONSTRAINTS_TASK_ERR: Final[str] = "no suitable node"
-_OSPARC_SERVICE_READY_LABEL_KEY: Final[DockerLabelKey] = parse_obj_as(
-    DockerLabelKey, "io.simcore.osparc-services-ready"
-)
-_OSPARC_SERVICES_READY_DATETIME_LABEL_KEY: Final[DockerLabelKey] = parse_obj_as(
-    DockerLabelKey, f"{_OSPARC_SERVICE_READY_LABEL_KEY}-last-changed"
+_OSPARC_SERVICE_READY_LABEL_KEY: Final[DockerLabelKey] = TypeAdapter(
+    DockerLabelKey
+).validate_python(
+    "io.simcore.osparc-services-ready",
 )
+_OSPARC_SERVICES_READY_DATETIME_LABEL_KEY: Final[DockerLabelKey] = TypeAdapter(
+    DockerLabelKey
+).validate_python(f"{_OSPARC_SERVICE_READY_LABEL_KEY}-last-changed")
 _OSPARC_SERVICE_READY_LABEL_KEYS: Final[list[DockerLabelKey]] = [
     _OSPARC_SERVICE_READY_LABEL_KEY,
     _OSPARC_SERVICES_READY_DATETIME_LABEL_KEY,
 ]
 
 
-_OSPARC_NODE_EMPTY_DATETIME_LABEL_KEY: Final[DockerLabelKey] = parse_obj_as(
-    DockerLabelKey, "io.simcore.osparc-node-found-empty"
-)
+_OSPARC_NODE_EMPTY_DATETIME_LABEL_KEY: Final[DockerLabelKey] = TypeAdapter(
+    DockerLabelKey
+).validate_python("io.simcore.osparc-node-found-empty")
 
-_OSPARC_NODE_TERMINATION_PROCESS_LABEL_KEY: Final[DockerLabelKey] = parse_obj_as(
-    DockerLabelKey, "io.simcore.osparc-node-termination-started"
-)
+_OSPARC_NODE_TERMINATION_PROCESS_LABEL_KEY: Final[DockerLabelKey] = TypeAdapter(
+    DockerLabelKey
+).validate_python("io.simcore.osparc-node-termination-started")
 
 
 async def get_monitored_nodes(
@@ -86,23 +88,21 @@ async def get_monitored_nodes(
     node_label_filters = [f"{label}=true" for label in node_labels] + [
         f"{label}" for label in _OSPARC_SERVICE_READY_LABEL_KEYS
     ]
-    return parse_obj_as(
-        list[Node],
-        await docker_client.nodes.list(filters={"node.label": node_label_filters}),
+    return TypeAdapter(list[Node]).validate_python(
+        await docker_client.nodes.list(filters={"node.label": node_label_filters})
     )
 
 
 async def get_worker_nodes(docker_client: AutoscalingDocker) -> list[Node]:
-    return parse_obj_as(
-        list[Node],
+    return TypeAdapter(list[Node]).validate_python(
         await docker_client.nodes.list(
             filters={
                 "role": ["worker"],
                 "node.label": [
                     f"{label}" for label in _OSPARC_SERVICE_READY_LABEL_KEYS
                 ],
             }
-        ),
+        )
     )
 
 
@@ -161,8 +161,8 @@ async def _associated_service_has_no_node_placement_contraints(
     docker_client: AutoscalingDocker, task: Task
 ) -> bool:
     assert task.ServiceID  # nosec
-    service_inspect = parse_obj_as(
-        Service, await docker_client.services.inspect(task.ServiceID)
+    service_inspect = TypeAdapter(Service).validate_python(
+        await docker_client.services.inspect(task.ServiceID)
     )
     assert service_inspect.Spec  # nosec
     assert service_inspect.Spec.TaskTemplate  # nosec
@@ -190,11 +190,9 @@ def _by_created_dt(task: Task) -> datetime.datetime:
     if task.CreatedAt:
         with suppress(ValueError):
             created_at = to_datetime(task.CreatedAt)
-            created_at_utc: datetime.datetime = created_at.replace(
-                tzinfo=datetime.timezone.utc
-            )
+            created_at_utc: datetime.datetime = created_at.replace(tzinfo=datetime.UTC)
             return created_at_utc
-    return datetime.datetime.now(datetime.timezone.utc)
+    return datetime.datetime.now(datetime.UTC)
 
 
 async def pending_service_tasks_with_insufficient_resources(
@@ -209,14 +207,13 @@ async def pending_service_tasks_with_insufficient_resources(
     - have an error message with "insufficient resources"
     - are not scheduled on any node
     """
-    tasks = parse_obj_as(
-        list[Task],
+    tasks = TypeAdapter(list[Task]).validate_python(
         await docker_client.tasks.list(
             filters={
                 "desired-state": "running",
                 "label": service_labels,
             }
-        ),
+        )
     )
 
     sorted_tasks = sorted(tasks, key=_by_created_dt)
@@ -264,7 +261,7 @@ async def compute_cluster_total_resources(nodes: list[Node]) -> Resources:
             }
         )
 
-    return Resources.parse_obj(dict(cluster_resources_counter))
+    return Resources.model_validate(dict(cluster_resources_counter))
 
 
 def get_max_resources_from_docker_task(task: Task) -> Resources:
@@ -285,16 +282,15 @@ def get_max_resources_from_docker_task(task: Task) -> Resources:
                 ),
             )
             / _NANO_CPU,
-            ram=parse_obj_as(
-                ByteSize,
+            ram=TypeAdapter(ByteSize).validate_python(
                 max(
                     task.Spec.Resources.Reservations
                     and task.Spec.Resources.Reservations.MemoryBytes
                     or 0,
                     task.Spec.Resources.Limits
                     and task.Spec.Resources.Limits.MemoryBytes
                     or 0,
-                ),
+                )
             ),
         )
     return Resources(cpus=0, ram=ByteSize(0))
@@ -305,8 +301,8 @@ async def get_task_instance_restriction(
 ) -> InstanceTypeType | None:
     with contextlib.suppress(ValidationError):
         assert task.ServiceID  # nosec
-        service_inspect = parse_obj_as(
-            Service, await docker_client.services.inspect(task.ServiceID)
+        service_inspect = TypeAdapter(Service).validate_python(
+            await docker_client.services.inspect(task.ServiceID)
         )
         assert service_inspect.Spec  # nosec
         assert service_inspect.Spec.TaskTemplate  # nosec
@@ -326,8 +322,8 @@ async def get_task_instance_restriction(
         )
         for constraint in service_placement_constraints:
             if constraint.startswith(node_label_to_find):
-                return parse_obj_as(
-                    InstanceTypeType, constraint.removeprefix(node_label_to_find)  # type: ignore[arg-type]
+                return TypeAdapter(InstanceTypeType).validate_python(
+                    constraint.removeprefix(node_label_to_find)
                 )
 
         return None
@@ -351,9 +347,8 @@ async def compute_node_used_resources(
     task_filters: dict[str, str | list[DockerLabelKey]] = {"node": node.ID}
     if service_labels is not None:
         task_filters |= {"label": service_labels}
-    all_tasks_on_node = parse_obj_as(
-        list[Task],
-        await docker_client.tasks.list(filters=task_filters),
+    all_tasks_on_node = TypeAdapter(list[Task]).validate_python(
+        await docker_client.tasks.list(filters=task_filters)
     )
     for task in all_tasks_on_node:
         assert task.Status  # nosec
@@ -370,7 +365,7 @@ async def compute_node_used_resources(
                     "cpus": task_reservations.get("NanoCPUs", 0) / _NANO_CPU,
                 }
             )
-    return Resources.parse_obj(dict(cluster_resources_counter))
+    return Resources.model_validate(dict(cluster_resources_counter))
 
 
 async def compute_cluster_used_resources(
@@ -380,11 +375,11 @@ async def compute_cluster_used_resources(
     list_of_used_resources = await logged_gather(
         *(compute_node_used_resources(docker_client, node) for node in nodes)
     )
-    counter = collections.Counter({k: 0 for k in Resources.__fields__})
+    counter = collections.Counter({k: 0 for k in Resources.model_fields})
     for result in list_of_used_resources:
         counter.update(result.dict())
 
-    return Resources.parse_obj(dict(counter))
+    return Resources.model_validate(dict(counter))
 
 
 _COMMAND_TIMEOUT_S = 10
@@ -446,10 +441,7 @@ def write_compose_file_command(
         },
     }
     compose_yaml = yaml.safe_dump(compose)
-    write_compose_file_cmd = " ".join(
-        ["echo", f'"{compose_yaml}"', ">", f"{_PRE_PULL_COMPOSE_PATH}"]
-    )
-    return write_compose_file_cmd
+    return " ".join(["echo", f'"{compose_yaml}"', ">", f"{_PRE_PULL_COMPOSE_PATH}"])
 
 
 def get_docker_pull_images_on_start_bash_command(
@@ -504,7 +496,7 @@ async def find_node_with_name(
     if not list_of_nodes:
         return None
     # note that there might be several nodes with a common_prefixed name. so now we want exact matching
-    parsed_list_of_nodes = parse_obj_as(list[Node], list_of_nodes)
+    parsed_list_of_nodes = TypeAdapter(list[Node]).validate_python(list_of_nodes)
     for node in parsed_list_of_nodes:
         assert node.Description  # nosec
         if node.Description.Hostname == name:
@@ -525,8 +517,8 @@ async def tag_node(
     ):
         assert node.ID  # nosec
 
-        latest_version_node = parse_obj_as(
-            Node, await docker_client.nodes.inspect(node_id=node.ID)
+        latest_version_node = TypeAdapter(Node).validate_python(
+            await docker_client.nodes.inspect(node_id=node.ID)
         )
         assert latest_version_node.Version  # nosec
         assert latest_version_node.Version.Index  # nosec
@@ -543,7 +535,9 @@ async def tag_node(
                 "Role": latest_version_node.Spec.Role.value,
             },
         )
-        return parse_obj_as(Node, await docker_client.nodes.inspect(node_id=node.ID))
+        return TypeAdapter(Node).validate_python(
+            await docker_client.nodes.inspect(node_id=node.ID)
+        )
 
 
 async def set_node_availability(

services/autoscaling/tests/unit/test_utils_rabbitmq.py

@@ -82,7 +82,7 @@ async def test_post_task_log_message(
     service_tasks = TypeAdapter(list[Task]).validate_python(
         await async_docker_client.tasks.list(
             filters={"service": service_with_labels.Spec.Name}
-        ),
+        )
     )
     assert service_tasks
     assert len(service_tasks) == 1
@@ -128,7 +128,7 @@ async def test_post_task_log_message_does_not_raise_if_service_has_no_labels(
     service_tasks = TypeAdapter(list[Task]).validate_python(
         await async_docker_client.tasks.list(
             filters={"service": service_without_labels.Spec.Name}
-        ),
+        )
     )
     assert service_tasks
     assert len(service_tasks) == 1
@@ -172,7 +172,7 @@ async def test_post_task_progress_message(
     service_tasks = TypeAdapter(list[Task]).validate_python(
         await async_docker_client.tasks.list(
             filters={"service": service_with_labels.Spec.Name}
-        ),
+        )
     )
     assert service_tasks
     assert len(service_tasks) == 1
@@ -218,7 +218,7 @@ async def test_post_task_progress_does_not_raise_if_service_has_no_labels(
     service_tasks = TypeAdapter(list[Task]).validate_python(
         await async_docker_client.tasks.list(
             filters={"service": service_without_labels.Spec.Name}
-        ),
+        )
     )
     assert service_tasks
     assert len(service_tasks) == 1