migrate to pydantic v2 (#344)

* fix: run `bump-pydantic nmdc_runtime` and apply

closes #339

addresses #343

* fix: @model_validator refactor

closes #343

nmdc_runtime/api/core/auth.py

@@ -35,7 +35,7 @@ class TokenExpires(BaseModel):
 class Token(BaseModel):
     access_token: str
     token_type: str
-    expires: Optional[TokenExpires]
+    expires: Optional[TokenExpires] = None
 
 
 class TokenData(BaseModel):

nmdc_runtime/api/models/capability.py

@@ -5,7 +5,7 @@
 
 
 class CapabilityBase(BaseModel):
-    name: Optional[str]
+    name: Optional[str] = None
     description: Optional[str] = None
 
 

nmdc_runtime/api/models/id.py

@@ -2,7 +2,8 @@
 from enum import Enum
 from typing import Union, Any, Optional, Literal
 
-from pydantic import BaseModel, constr, PositiveInt, root_validator
+from pydantic import model_validator, StringConstraints, BaseModel, PositiveInt
+from typing_extensions import Annotated
 
 # NO i, l, o or u.
 base32_letters = "abcdefghjkmnpqrstvwxyz"
@@ -22,11 +23,11 @@
 _base_object_name = f"{_naa}:{_shoulder}{_blade}"
 pattern_base_object_name = re.compile(_base_object_name)
 
-Naa = constr(pattern=_naa)
-Shoulder = constr(pattern=rf"^{_shoulder}$", min_length=2)
-Blade = constr(pattern=_blade, min_length=4)
-AssignedBaseName = constr(pattern=_assigned_base_name)
-BaseObjectName = constr(pattern=_base_object_name)
+Naa = Annotated[str, StringConstraints(pattern=_naa)]
+Shoulder = Annotated[str, StringConstraints(pattern=rf"^{_shoulder}$", min_length=2)]
+Blade = Annotated[str, StringConstraints(pattern=_blade, min_length=4)]
+AssignedBaseName = Annotated[str, StringConstraints(pattern=_assigned_base_name)]
+BaseObjectName = Annotated[str, StringConstraints(pattern=_base_object_name)]
 
 NameAssigningAuthority = Literal[tuple(NAA_VALUES)]
 
@@ -71,18 +72,18 @@ class IdBindingOp(str, Enum):
 class IdBindingRequest(BaseModel):
     i: BaseObjectName
     o: IdBindingOp = IdBindingOp.set
-    a: Optional[str]
-    v: Any
+    a: Optional[str] = None
+    v: Any = None
 
-    @root_validator(skip_on_failure=True)
+    @model_validator(mode="before")
     def set_or_add_needs_value(cls, values):
         op = values.get("o")
         if op in (IdBindingOp.set, IdBindingOp.addToSet):
             if "v" not in values:
                 raise ValueError("{'set','add'} operations needs value 'v'.")
         return values
 
-    @root_validator(skip_on_failure=True)
+    @model_validator(mode="before")
     def set_or_add_or_rm_needs_attribute(cls, values):
         op = values.get("o")
         if op in (IdBindingOp.set, IdBindingOp.addToSet, IdBindingOp.rm):

nmdc_runtime/api/models/job.py

@@ -9,7 +9,7 @@
 
 class JobBase(BaseModel):
     workflow: Workflow
-    name: Optional[str]
+    name: Optional[str] = None
     description: Optional[str] = None
 
 
@@ -20,7 +20,7 @@ class JobClaim(BaseModel):
 
 class Job(JobBase):
     id: str
-    created_at: Optional[datetime.datetime]
+    created_at: Optional[datetime.datetime] = None
     config: Dict[str, Any]
     claims: List[JobClaim] = []
 

nmdc_runtime/api/models/metadata.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Extra
+from pydantic import ConfigDict, BaseModel
 
 
 class ChangesheetIn(BaseModel):
@@ -8,5 +8,4 @@ class ChangesheetIn(BaseModel):
 
 
 class Doc(BaseModel):
-    class Config:
-        extra = Extra.allow
+    model_config = ConfigDict(extra="allow")

nmdc_runtime/api/models/object.py

@@ -5,14 +5,15 @@
 from typing import Optional, List, Dict
 
 from pydantic import (
+    field_validator,
+    model_validator,
+    Field,
+    StringConstraints,
     BaseModel,
     AnyUrl,
-    constr,
-    conint,
     HttpUrl,
-    root_validator,
-    validator,
 )
+from typing_extensions import Annotated
 
 
 class AccessMethodType(str, Enum):
@@ -27,17 +28,17 @@ class AccessMethodType(str, Enum):
 
 
 class AccessURL(BaseModel):
-    headers: Optional[Dict[str, str]]
+    headers: Optional[Dict[str, str]] = None
     url: AnyUrl
 
 
 class AccessMethod(BaseModel):
-    access_id: Optional[constr(min_length=1)]
-    access_url: Optional[AccessURL]
-    region: Optional[str]
+    access_id: Optional[Annotated[str, StringConstraints(min_length=1)]] = None
+    access_url: Optional[AccessURL] = None
+    region: Optional[str] = None
     type: AccessMethodType = AccessMethodType.https
 
-    @root_validator(skip_on_failure=True)
+    @model_validator(mode="before")
     def at_least_one_of_access_id_and_url(cls, values):
         access_id, access_url = values.get("access_id"), values.get("access_url")
         if access_id is None and access_url is None:
@@ -47,27 +48,30 @@ def at_least_one_of_access_id_and_url(cls, values):
         return values
 
 
-ChecksumType = constr(
-    pattern=rf"(?P<checksumtype>({'|'.join(sorted(hashlib.algorithms_guaranteed))}))"
-)
+ChecksumType = Annotated[
+    str,
+    StringConstraints(
+        pattern=rf"(?P<checksumtype>({'|'.join(sorted(hashlib.algorithms_guaranteed))}))"
+    ),
+]
 
 
 class Checksum(BaseModel):
-    checksum: constr(min_length=1)
+    checksum: Annotated[str, StringConstraints(min_length=1)]
     type: ChecksumType
 
 
-DrsId = constr(pattern=r"^[A-Za-z0-9._~\-]+$")
-PortableFilename = constr(pattern=r"^[A-Za-z0-9._\-]+$")
+DrsId = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9._~\-]+$")]
+PortableFilename = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9._\-]+$")]
 
 
 class ContentsObject(BaseModel):
-    contents: Optional[List["ContentsObject"]]
-    drs_uri: Optional[List[AnyUrl]]
-    id: Optional[DrsId]
+    contents: Optional[List["ContentsObject"]] = None
+    drs_uri: Optional[List[AnyUrl]] = None
+    id: Optional[DrsId] = None
     name: PortableFilename
 
-    @root_validator(skip_on_failure=True)
+    @model_validator(mode="before")
     def no_contents_means_single_blob(cls, values):
         contents, id_ = values.get("contents"), values.get("id")
         if contents is None and id_ is None:
@@ -77,32 +81,32 @@ def no_contents_means_single_blob(cls, values):
 
 ContentsObject.update_forward_refs()
 
-Mimetype = constr(pattern=r"^\w+/[-+.\w]+$")
-SizeInBytes = conint(strict=True, ge=0)
+Mimetype = Annotated[str, StringConstraints(pattern=r"^\w+/[-+.\w]+$")]
+SizeInBytes = Annotated[int, Field(strict=True, ge=0)]
 
 
 class Error(BaseModel):
-    msg: Optional[str]
+    msg: Optional[str] = None
     status_code: http.HTTPStatus
 
 
 class DrsObjectBase(BaseModel):
-    aliases: Optional[List[str]]
+    aliases: Optional[List[str]] = None
     description: Optional[str] = None
-    mime_type: Optional[Mimetype]
-    name: Optional[PortableFilename]
+    mime_type: Optional[Mimetype] = None
+    name: Optional[PortableFilename] = None
 
 
 class DrsObjectIn(DrsObjectBase):
-    access_methods: Optional[List[AccessMethod]]
+    access_methods: Optional[List[AccessMethod]] = None
     checksums: List[Checksum]
-    contents: Optional[List[ContentsObject]]
+    contents: Optional[List[ContentsObject]] = None
     created_time: datetime.datetime
     size: SizeInBytes
-    updated_time: Optional[datetime.datetime]
-    version: Optional[str]
+    updated_time: Optional[datetime.datetime] = None
+    version: Optional[str] = None
 
-    @root_validator(skip_on_failure=True)
+    @model_validator(mode="before")
     def no_contents_means_single_blob(cls, values):
         contents, access_methods = values.get("contents"), values.get("access_methods")
         if contents is None and access_methods is None:
@@ -111,7 +115,8 @@ def no_contents_means_single_blob(cls, values):
             )
         return values
 
-    @validator("checksums")
+    @field_validator("checksums")
+    @classmethod
     def at_least_one_checksum(cls, v):
         if not len(v) >= 1:
             raise ValueError("At least one checksum requried")
@@ -123,7 +128,7 @@ class DrsObject(DrsObjectIn):
     self_uri: AnyUrl
 
 
-Seconds = conint(strict=True, gt=0)
+Seconds = Annotated[int, Field(strict=True, gt=0)]
 
 
 class ObjectPresignedUrl(BaseModel):
@@ -137,14 +142,14 @@ class DrsObjectOutBase(DrsObjectBase):
     id: DrsId
     self_uri: AnyUrl
     size: SizeInBytes
-    updated_time: Optional[datetime.datetime]
-    version: Optional[str]
+    updated_time: Optional[datetime.datetime] = None
+    version: Optional[str] = None
 
 
 class DrsObjectBlobOut(DrsObjectOutBase):
     access_methods: List[AccessMethod]
 
 
 class DrsObjectBundleOut(DrsObjectOutBase):
-    access_methods: Optional[List[AccessMethod]]
+    access_methods: Optional[List[AccessMethod]] = None
     contents: List[ContentsObject]

nmdc_runtime/api/models/object_type.py

@@ -7,7 +7,7 @@
 
 
 class ObjectTypeBase(BaseModel):
-    name: Optional[str]
+    name: Optional[str] = None
     description: Optional[str] = None
 
 
@@ -17,4 +17,4 @@ class ObjectType(ObjectTypeBase):
 
 
 class DrsObjectWithTypes(DrsObject):
-    types: Optional[List[str]]
+    types: Optional[List[str]] = None

nmdc_runtime/api/models/operation.py

@@ -1,44 +1,44 @@
 import datetime
 from typing import Generic, TypeVar, Optional, List, Any, Union
 
-from pydantic import BaseModel, HttpUrl, constr
-from pydantic.generics import GenericModel
+from pydantic import StringConstraints, BaseModel, HttpUrl
 
 from nmdc_runtime.api.models.util import ResultT
+from typing_extensions import Annotated
 
 MetadataT = TypeVar("MetadataT")
 
 
-PythonImportPath = constr(pattern=r"^[A-Za-z0-9_.]+$")
+PythonImportPath = Annotated[str, StringConstraints(pattern=r"^[A-Za-z0-9_.]+$")]
 
 
 class OperationError(BaseModel):
     code: str
     message: str
-    details: Any
+    details: Any = None
 
 
-class Operation(GenericModel, Generic[ResultT, MetadataT]):
+class Operation(BaseModel, Generic[ResultT, MetadataT]):
     id: str
     done: bool = False
     expire_time: datetime.datetime
-    result: Optional[Union[ResultT, OperationError]]
-    metadata: Optional[MetadataT]
+    result: Optional[Union[ResultT, OperationError]] = None
+    metadata: Optional[MetadataT] = None
 
 
-class UpdateOperationRequest(GenericModel, Generic[ResultT, MetadataT]):
+class UpdateOperationRequest(BaseModel, Generic[ResultT, MetadataT]):
     done: bool = False
-    result: Optional[Union[ResultT, OperationError]]
+    result: Optional[Union[ResultT, OperationError]] = None
     metadata: Optional[MetadataT] = {}
 
 
-class ListOperationsResponse(GenericModel, Generic[ResultT, MetadataT]):
+class ListOperationsResponse(BaseModel, Generic[ResultT, MetadataT]):
     resources: List[Operation[ResultT, MetadataT]]
-    next_page_token: Optional[str]
+    next_page_token: Optional[str] = None
 
 
 class Result(BaseModel):
-    model: Optional[PythonImportPath]
+    model: Optional[PythonImportPath] = None
 
 
 class EmptyResult(Result):
@@ -47,7 +47,7 @@ class EmptyResult(Result):
 
 class Metadata(BaseModel):
     # XXX alternative: set model field using __class__ on __init__()?
-    model: Optional[PythonImportPath]
+    model: Optional[PythonImportPath] = None
 
 
 class PausedOrNot(Metadata):