♻️ Maintenance: Migrate more Pydantic models to Annotated types

.github/prompts/pydantic-annotated-fields.prompt.md

@@ -13,7 +13,7 @@ Follow these guidelines:
 4. Add the import: `from common_library.basic_types import DEFAULT_FACTORY` if it's not already present.
 5. If `Field()` has no parameters (empty), don't use Annotated at all. Just use: `field_name: field_type = default_value`.
 6. Leave any model validations, `model_config` settings, and `field_validators` untouched.
-
+7. Must keep the original Field descriptions and validation parameters intact (except for the `default` parameter).
 
 ## Examples
 

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

@@ -5,6 +5,7 @@
 from typing import Annotated, Final, TypeAlias
 
 import sh  # type: ignore[import-untyped]
+from common_library.basic_types import DEFAULT_FACTORY
 from models_library.docker import DockerGenericTag
 from pydantic import (
     BaseModel,
@@ -143,23 +144,32 @@ class EC2InstanceConfig:
 
 class EC2InstanceBootSpecific(BaseModel):
     ami_id: AMIIdStr
-    custom_boot_scripts: list[CommandStr] = Field(
-        default_factory=list,
-        description="script(s) to run on EC2 instance startup (be careful!), "
-        "each entry is run one after the other using '&&' operator",
-    )
-    pre_pull_images: list[DockerGenericTag] = Field(
-        default_factory=list,
-        description="a list of docker image/tags to pull on instance cold start",
-    )
-    pre_pull_images_cron_interval: datetime.timedelta = Field(
-        default=datetime.timedelta(minutes=30),
-        description="time interval between pulls of images (minimum is 1 minute) "
-        "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formating)",
-    )
-    buffer_count: NonNegativeInt = Field(
-        default=0, description="number of buffer EC2s to keep (defaults to 0)"
-    )
+    custom_boot_scripts: Annotated[
+        list[CommandStr],
+        Field(
+            default_factory=list,
+            description="script(s) to run on EC2 instance startup (be careful!), "
+            "each entry is run one after the other using '&&' operator",
+        ),
+    ] = DEFAULT_FACTORY
+    pre_pull_images: Annotated[
+        list[DockerGenericTag],
+        Field(
+            default_factory=list,
+            description="a list of docker image/tags to pull on instance cold start",
+        ),
+    ] = DEFAULT_FACTORY
+    pre_pull_images_cron_interval: Annotated[
+        datetime.timedelta,
+        Field(
+            description="time interval between pulls of images (minimum is 1 minute) "
+            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formating)",
+        ),
+    ] = datetime.timedelta(minutes=30)
+    buffer_count: Annotated[
+        NonNegativeInt,
+        Field(description="number of buffer EC2s to keep (defaults to 0)"),
+    ] = 0
 
     @field_validator("custom_boot_scripts")
     @classmethod

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

@@ -1,6 +1,6 @@
 import datetime
 from pathlib import Path
-from typing import TypeAlias, cast
+from typing import Annotated, TypeAlias, cast
 
 from models_library.api_schemas_storage.storage_schemas import ETag
 from models_library.basic_types import SHA256Str
@@ -54,9 +54,10 @@ def as_path(self) -> Path:
 
 class S3DirectoryMetaData(BaseModel, frozen=True):
     prefix: S3ObjectPrefix
-    size: ByteSize | None = Field(
-        ..., description="Size of the directory if computed, None if unknown"
-    )
+    size: Annotated[
+        ByteSize | None,
+        Field(description="Size of the directory if computed, None if unknown"),
+    ]
 
     def as_path(self) -> Path:
         return self.prefix

packages/common-library/tests/test_pydantic_fields_extension.py

@@ -1,13 +1,14 @@
-from typing import Any, Callable, Literal
+from collections.abc import Callable
+from typing import Any, Literal
 
 import pytest
 from common_library.pydantic_fields_extension import get_type, is_literal, is_nullable
-from pydantic import BaseModel, Field
+from pydantic import BaseModel
 
 
 class MyModel(BaseModel):
     a: int
-    b: float | None = Field(...)
+    b: float | None
     c: str = "bla"
     d: bool | None = None
     e: Literal["bla"]

packages/common-library/tests/test_pydantic_validators.py

@@ -7,7 +7,7 @@
     validate_numeric_string_as_timedelta,
 )
 from faker import Faker
-from pydantic import BeforeValidator, Field
+from pydantic import BeforeValidator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 
@@ -17,7 +17,7 @@ class Settings(BaseSettings):
         APP_NAME: str
         REQUEST_TIMEOUT: Annotated[
             timedelta, BeforeValidator(_validate_legacy_timedelta_str)
-        ] = Field(default=timedelta(hours=1))
+        ] = timedelta(hours=1)
 
         model_config = SettingsConfigDict()
 
@@ -45,7 +45,7 @@ def test_validate_timedelta_in_legacy_mode(
 ):
     class Settings(BaseSettings):
         APP_NAME: str
-        REQUEST_TIMEOUT: timedelta = Field(default=timedelta(seconds=40))
+        REQUEST_TIMEOUT: timedelta = timedelta(seconds=40)
 
         _validate_request_timeout = validate_numeric_string_as_timedelta(
             "REQUEST_TIMEOUT"

packages/dask-task-models-library/src/dask_task_models_library/container_tasks/io.py

@@ -75,13 +75,15 @@ def _update_json_schema_extra(schema: JsonDict) -> None:
 
 class FileUrl(BaseModel):
     url: AnyUrl
-    file_mapping: str | None = Field(
-        default=None,
-        description="Local file relpath name (if given), otherwise it takes the url filename",
-    )
-    file_mime_type: str | None = Field(
-        default=None, description="the file MIME type", pattern=MIME_TYPE_RE
-    )
+    file_mapping: Annotated[
+        str | None,
+        Field(
+            description="Local file relpath name (if given), otherwise it takes the url filename"
+        ),
+    ] = None
+    file_mime_type: Annotated[
+        str | None, Field(description="the file MIME type", pattern=MIME_TYPE_RE)
+    ] = None
 
     @staticmethod
     def _update_json_schema_extra(schema: JsonDict) -> None:

packages/models-library/src/models_library/api_schemas__common/errors.py

@@ -1,18 +1,16 @@
 import http
-from typing import Any
+from typing import Annotated, Any
 
 from pydantic import BaseModel, Field
 
 from ..basic_types import IDStr
 
 
 class DefaultApiError(BaseModel):
-    name: IDStr = Field(
-        ...,
-        description="Error identifier as a code or a name. "
-        "Mainly for machine-machine communication purposes.",
+    name: Annotated[IDStr, Field(description="Exception's class name")]
+    detail: Annotated[Any | None, Field(description="Human readable error message")] = (
+        None
     )
-    detail: Any | None = Field(default=None, description="Human readable error message")
 
     @classmethod
     def from_status_code(

packages/models-library/src/models_library/api_schemas__common/meta.py

@@ -1,3 +1,5 @@
+from typing import Annotated
+
 from pydantic import BaseModel, ConfigDict, Field
 
 from ..basic_types import VersionStr
@@ -6,9 +8,10 @@
 class BaseMeta(BaseModel):
     name: str
     version: VersionStr
-    released: dict[str, VersionStr] | None = Field(
-        default=None, description="Maps every route's path tag with a released version"
-    )
+    released: Annotated[
+        dict[str, VersionStr] | None,
+        Field(description="Maps every route's path tag with a released version"),
+    ] = None
 
     model_config = ConfigDict(
         json_schema_extra={

packages/models-library/src/models_library/api_schemas_api_server/api_keys.py

@@ -1,3 +1,5 @@
+from typing import Annotated
+
 from pydantic import BaseModel, ConfigDict, Field, SecretStr
 
 
@@ -10,7 +12,7 @@ class ApiKeyInDB(BaseModel):
     api_key: str
     api_secret: str
 
-    id_: int = Field(0, alias="id")
+    id_: Annotated[int, Field(alias="id")] = 0
     display_name: str
     user_id: int
     product_name: str

packages/models-library/src/models_library/api_schemas_catalog/services_specifications.py

@@ -1,18 +1,23 @@
+from typing import Annotated
+
 from pydantic import BaseModel, Field
 
 from ..generated_models.docker_rest_api import ServiceSpec as DockerServiceSpec
 
 
 class ServiceSpecifications(BaseModel):
-    sidecar: DockerServiceSpec | None = Field(
-        default=None,
-        description="schedule-time specifications for the service sidecar (follows Docker Service creation API, see https://docs.docker.com/engine/api/v1.25/#operation/ServiceCreate)",
-    )
-    service: DockerServiceSpec | None = Field(
-        default=None,
-        description="schedule-time specifications specifications for the service (follows Docker Service creation API (specifically only the Resources part), see https://docs.docker.com/engine/api/v1.41/#tag/Service/operation/ServiceCreate",
-    )
+    sidecar: Annotated[
+        DockerServiceSpec | None,
+        Field(
+            description="schedule-time specifications for the service sidecar (follows Docker Service creation API, see https://docs.docker.com/engine/api/v1.25/#operation/ServiceCreate)",
+        ),
+    ] = None
+    service: Annotated[
+        DockerServiceSpec | None,
+        Field(
+            description="schedule-time specifications specifications for the service (follows Docker Service creation API (specifically only the Resources part), see https://docs.docker.com/engine/api/v1.41/#tag/Service/operation/ServiceCreate",
+        ),
+    ] = None
 
 
-class ServiceSpecificationsGet(ServiceSpecifications):
-    ...
+class ServiceSpecificationsGet(ServiceSpecifications): ...

packages/models-library/src/models_library/api_schemas_clusters_keeper/clusters.py

@@ -1,5 +1,6 @@
 import datetime
 from enum import auto
+from typing import Annotated
 
 from pydantic import AnyUrl, BaseModel, Field
 
@@ -17,7 +18,7 @@ class ClusterState(StrAutoEnum):
 
 class OnDemandCluster(BaseModel):
     endpoint: AnyUrl
-    authentication: ClusterAuthentication = Field(discriminator="type")
+    authentication: Annotated[ClusterAuthentication, Field(discriminator="type")]
     state: ClusterState
     user_id: UserID
     wallet_id: WalletID | None

packages/models-library/src/models_library/api_schemas_directorv2/clusters.py

@@ -20,10 +20,10 @@ class TaskCounts(BaseModel):
 
 
 class WorkerMetrics(BaseModel):
-    cpu: float = Field(..., description="consumed % of cpus")
-    memory: ByteSize = Field(..., description="consumed memory")
-    num_fds: int = Field(..., description="consumed file descriptors")
-    task_counts: TaskCounts = Field(..., description="task details")
+    cpu: Annotated[float, Field(description="consumed % of cpus")]
+    memory: Annotated[ByteSize, Field(description="consumed memory")]
+    num_fds: Annotated[int, Field(description="consumed file descriptors")]
+    task_counts: Annotated[TaskCounts, Field(description="task details")]
 
 
 AvailableResources: TypeAlias = DictModel[str, PositiveFloat]
@@ -54,7 +54,7 @@ class Worker(BaseModel):
 
 
 class Scheduler(BaseModel):
-    status: str = Field(..., description="The running status of the scheduler")
+    status: Annotated[str, Field(description="The running status of the scheduler")]
     workers: Annotated[WorkersDict | None, Field(default_factory=dict)]
 
     @field_validator("workers", mode="before")
@@ -66,10 +66,5 @@ def ensure_workers_is_empty_dict(cls, v):
 
 
 class ClusterDetails(BaseModel):
-    scheduler: Scheduler = Field(
-        ...,
-        description="This contains dask scheduler information given by the underlying dask library",
-    )
-    dashboard_link: AnyUrl = Field(
-        ..., description="Link to this scheduler's dashboard"
-    )
+    scheduler: Annotated[Scheduler, Field(description="scheduler information")]
+    dashboard_link: Annotated[AnyUrl, Field(description="Link to the dask dashboard")]

packages/models-library/src/models_library/api_schemas_directorv2/errors.py

@@ -1,14 +1,16 @@
+from typing import Annotated
+
 from pydantic import BaseModel, Field
 
 
 class Error(BaseModel):
-    code: str | None = Field(None, description="Server Exception")
+    code: Annotated[str | None, Field(description="Server Exception")] = None
 
 
 class ErrorType(BaseModel):
-    message: str = Field(..., description="Error message")
+    message: Annotated[str, Field(description="Error message")]
+    status: Annotated[int, Field(description="Error code")]
     errors: list[Error] | None = None
-    status: int = Field(..., description="Error code")
 
 
 class ErrorEnveloped(BaseModel):

packages/models-library/src/models_library/api_schemas_long_running_tasks/base.py

@@ -18,9 +18,9 @@ class TaskProgress(BaseModel):
     defined as a float bound between 0.0 and 1.0
     """
 
-    task_id: TaskId | None = Field(default=None)
-    message: ProgressMessage = Field(default="")
-    percent: ProgressPercent = Field(default=0.0)
+    task_id: TaskId | None = None
+    message: ProgressMessage = ""
+    percent: ProgressPercent = 0.0
 
     @validate_call
     def update(