Pydantic upgrade

api/common/settings.py

@@ -1,4 +1,5 @@
-from pydantic import BaseSettings, EmailStr, DirectoryPath
+from pydantic import EmailStr, DirectoryPath, Field
+from pydantic_settings import BaseSettings
 
 class APISettings(BaseSettings):
     max_size: int = 30000000 # 30MB
@@ -10,8 +11,8 @@ class APISettings(BaseSettings):
     smtp_host: str
     smtp_port: int
     smtp_starttls: bool
-    smtp_username: str | None
-    smtp_password: str | None
+    smtp_username: str | None = Field(None)
+    smtp_password: str | None = Field(None)
     smtp_from: EmailStr
 
     class Config:

api/models/collections.py

@@ -1,26 +1,28 @@
-from datetime import datetime
-from typing import Optional
 from enum import Enum
+from typing import Annotated
+from datetime import datetime
 
-from pydantic import BaseModel, root_validator, constr
+from pydantic import BaseModel, StringConstraints, computed_field
+from pydantic_mongo import PydanticObjectId
 
-from .default import DefaultModel, PyObjectId
+from .default import DefaultModel
 
 class Collection(DefaultModel):
-    created_on: Optional[datetime]
-    owner: Optional[str]
+    @computed_field
+    @property
+    def created_on(self) -> datetime | None:
+        if not self.id:
+            return None
+        return self.id.generation_time
+    owner: str | None = None
     is_private: bool
-    description: constr(min_length=1, max_length=255)
+    description: Annotated[str, StringConstraints(min_length=1, max_length=255)] 
 
-    @root_validator
-    def get_created_date(cls, values) -> dict:
-        values["created_on"] = values["id"].generation_time
-        return values
 
 class NewCollection(BaseModel):
     is_private: bool
-    description: constr(min_length=1, max_length=255)
-    image_ids: list[PyObjectId] = []
+    description: Annotated[str, StringConstraints(min_length=1, max_length=255)] 
+    image_ids: list[PydanticObjectId] = []
 
 class EditableCollectionInformation(str, Enum):
     description = "description"

api/models/default.py

@@ -1,27 +1,8 @@
-from bson.objectid import ObjectId
-from pydantic import BaseModel, Field
-
-
-class PyObjectId(ObjectId):
-    @classmethod
-    def __get_validators__(cls):
-        yield cls.validate
-
-    @classmethod
-    def validate(cls, v):
-        if not ObjectId.is_valid(v):
-            raise ValueError("Invalid objectid")
-        return ObjectId(v)
-
-    @classmethod
-    def __modify_schema__(cls, field_schema):
-        field_schema.update(type="string")
-
+from pydantic import BaseModel, Field, ConfigDict
+from pydantic_mongo import PydanticObjectId
+from typing import Optional, Annotated
 
 class DefaultModel(BaseModel):
-    id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
+    id: Annotated[Optional[PydanticObjectId], Field(alias="_id")] = None
 
-    class Config:
-        allow_population_by_field_name = True
-        arbitrary_types_allowed = True
-        json_encoders = {ObjectId: str}
+    model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)

api/models/images.py

@@ -1,36 +1,38 @@
-from datetime import datetime
 from enum import Enum, IntEnum
 from uuid import UUID
+from typing import Annotated
+from datetime import datetime
 
-from pydantic import BaseModel, Field, constr, root_validator
+from pydantic import BaseModel, computed_field, StringConstraints
+from pydantic_mongo import PydanticObjectId
 
-from .default import DefaultModel, PyObjectId
+from .default import DefaultModel
 
 class ImageMetadata(BaseModel):
-    description: constr(min_length=1, max_length=255)
+    description: Annotated[str, StringConstraints(min_length=1, max_length=255)] 
     width: int
     height: int
-    real_content_type: str | None
+    real_content_type: str | None = None
 
 
 class File(BaseModel):
     content_type: str
     type_extension: str
-    salt: bytes | None
-    nonce: bytes | None
-    tag: bytes | None
+    salt: bytes | None = None
+    nonce: bytes | None = None
+    tag: bytes | None = None
 
 
 class Thumbnail(BaseModel):
-    is_computing: bool = Field(default_factory=lambda: False)
-    is_unavailable: bool = Field(default_factory=lambda: False)
+    is_computing: bool = False
+    is_unavailable: bool = False
 
 
 class ImageMetadataContainer(BaseModel):
-    salt: bytes | None
-    nonce: bytes | None
+    salt: bytes | None = None
+    nonce: bytes | None = None
     data: bytes | ImageMetadata
-    tag: bytes | None
+    tag: bytes | None = None
 
 
 class LockVersion(IntEnum):
@@ -39,39 +41,38 @@ class LockVersion(IntEnum):
 
 class Lock(BaseModel):
     is_locked: bool
-    version: LockVersion | None
-    upgradable: bool | None
+    version: LockVersion | None = None
 
-    @root_validator
-    def get_upgradable(cls, values) -> dict:
-        if values["version"]:
-            values["upgradable"] = values["version"] < LockVersion.aes128gcm_argon2
-        return values
+    @computed_field
+    @property
+    def upgradable(self) -> bool | None:
+        if self.version:
+            return self.version < LockVersion.aes128gcm_argon2
 
 
 class Image(DefaultModel):
-    created_on: datetime | None
-    owner: str | None
+    @computed_field
+    @property
+    def created_on(self) -> datetime | None:
+        if not self.id:
+            return None
+        return self.id.generation_time
+    owner: str | None = None
     is_private: bool
     lock: Lock
     file: File
-    thumbnail: Thumbnail | None
+    thumbnail: Thumbnail | None = None
     metadata: ImageMetadataContainer
 
-    @root_validator
-    def get_created_on(cls, values) -> dict:
-        values["created_on"] = values["id"].generation_time
-        return values
-
 class ImageInDB(Image):
-    collections: list[PyObjectId] = []
-    ownerless_key: UUID | None
+    collections: list[PydanticObjectId] = []
+    ownerless_key: UUID | None = None
 
 class ImageUpload(BaseModel):
-    description: constr(min_length=1, max_length=255)
+    description: Annotated[str, StringConstraints(min_length=1, max_length=255)] 
     is_private: bool
     is_locked: bool
-    lock_key: constr(min_length=3) | None
+    lock_key: Annotated[str, StringConstraints(min_length=3)] | None
 
 class EditableImageInformation(str, Enum):
     is_private = "is_private"

api/models/pagination.py

@@ -1,8 +1,10 @@
-from pydantic import BaseModel, conint
+from typing import Annotated
+
+from pydantic import BaseModel, Field
+from pydantic_mongo import PydanticObjectId
 
-from .default import PyObjectId
 
 class Pagination(BaseModel):
-    query: str | None
-    last_id: PyObjectId | None
-    limit: conint(ge=1, le=15) = 3
+    query: str | None = None
+    last_id: PydanticObjectId | None = None
+    limit: Annotated[int, Field(ge=1, le=15)]  = 3

api/models/users.py

@@ -3,16 +3,15 @@
 from datetime import datetime, timezone
 from enum import Enum
 
-from pydantic import BaseModel, Field, constr, EmailStr
+from pydantic import BaseModel, Field, constr, EmailStr, ConfigDict
 
 
 class User(BaseModel):
     username: constr(strip_whitespace=True, min_length=3) = Field(..., alias="_id")
-    email: EmailStr | None
+    email: EmailStr | None = None
     created_on: datetime = Field(default_factory=lambda: datetime.now(timezone.utc).replace(microsecond=0))
 
-    class Config:
-        allow_population_by_field_name = True
+    model_config = ConfigDict(populate_by_name=True)
 
 class UserInDB(User):
     password: str
@@ -26,5 +25,4 @@ class PasswordReset(BaseModel):
     code: str = Field(default_factory=lambda: ''.join(choice(digits) for x in range(6)))
     created_on: datetime = Field(default_factory=lambda: datetime.now(timezone.utc).replace(microsecond=0))
 
-    class Config:
-        allow_population_by_field_name = True
+    model_config = ConfigDict(populate_by_name=True)

api/routers/collections.py

@@ -1,26 +1,27 @@
 from secrets import compare_digest
 
-from fastapi import APIRouter, Body, Depends, HTTPException, Request, status
+from fastapi import APIRouter, Body, Depends, HTTPException, Request, status, Path
+from typing import Annotated
 from fastapi.responses import HTMLResponse
 from pymongo import DESCENDING
+from pydantic_mongo import PydanticObjectId
 
 from ..common.db import db_collections, db_images
 from ..common.security import get_optional_user, get_user
 from ..common.templates import templates
 from ..models.collections import (Collection, EditableCollectionInformation,
                                   NewCollection)
-from ..models.default import PyObjectId
 from ..models.images import Image, ImageInDB
 from ..models.pagination import Pagination
 from ..models.users import User
 
 
-def add_images(collection_id: PyObjectId, image_ids: list[PyObjectId], user: User | None):
+def add_images(collection_id: PydanticObjectId, image_ids: list[PydanticObjectId], user: User | None):
     for image_id in image_ids:
         image_dict = db_images.find_one({"_id": image_id})
         if not image_dict:
             raise HTTPException(status.HTTP_404_NOT_FOUND, detail=f"The image '{image_id}' does not exist.")
-        image = ImageInDB.parse_obj(image_dict)
+        image = ImageInDB.model_validate(image_dict)
         if image.is_private and not compare_digest(image.owner, user.username):
             raise HTTPException(status.HTTP_403_FORBIDDEN, detail=f"You don't have permission to access the image '{image_id}'.")
         if image_id in image.collections:
@@ -46,16 +47,16 @@ def new_collection(
         is_private=new_collection.is_private,
         description=new_collection.description
     )
-    db_collections.insert_one(collection.dict(by_alias=True, exclude={"created_on"}, exclude_none=True))
+    db_collections.insert_one(collection.model_dump(by_alias=True, exclude={"created_on"}, exclude_none=True))
     # Validate images list.
     add_images(collection.id, new_collection.image_ids, user)
     return collection
 
-def get_collection_in_db(id: PyObjectId, user: User | None) -> Collection:
+def get_collection_in_db(id: PydanticObjectId, user: User | None) -> Collection:
     collection_dict = db_collections.find_one({"_id": id})
     if not collection_dict:
         raise HTTPException(status.HTTP_404_NOT_FOUND)
-    collection = Collection.parse_obj(collection_dict)
+    collection = Collection.model_validate(collection_dict)
     if collection.is_private and (not user or not compare_digest(user.username, collection.owner)):
         raise HTTPException(status.HTTP_403_FORBIDDEN)
     return collection
@@ -67,7 +68,7 @@ def get_collection_in_db(id: PyObjectId, user: User | None) -> Collection:
     response_model_exclude_none=True
 )
 def get_collection(
-    id: PyObjectId,
+    id: Annotated[PydanticObjectId, Path(...)],
     user: User | None = Depends(get_optional_user)
 ):
     return get_collection_in_db(id, user)
@@ -77,9 +78,9 @@ def get_collection(
     status_code=status.HTTP_204_NO_CONTENT
 )
 def edit_collection(
-    id: PyObjectId,
+    id: PydanticObjectId,
     change: EditableCollectionInformation = Body(...),
-    to: bool | str | list[PyObjectId] = Body(...),
+    to: bool | str | list[PydanticObjectId] = Body(...),
     user: User = Depends(get_user)
 ):
     get_collection_in_db(id, user)
@@ -115,7 +116,7 @@ def edit_collection(
     status_code=status.HTTP_204_NO_CONTENT
 )
 def delete_collection(
-    id: PyObjectId,
+    id: PydanticObjectId,
     user: User = Depends(get_user)
 ):
     get_collection_in_db(id, user)
@@ -130,14 +131,14 @@ def delete_collection(
     response_class=HTMLResponse
 )
 def get_collection_embed(
-    id: PyObjectId,
+    id: PydanticObjectId,
     request: Request
 ):
     return templates.TemplateResponse("embed-collection.html", {
         "request": request,
         "collection": get_collection_in_db(id, None),
         "images": list(map( 
-            lambda i: Image.parse_obj(i),
+            lambda i: Image.model_validate(i),
             db_images.find({
                 "collections": id,
                 "is_private": False
@@ -152,7 +153,7 @@ def get_collection_embed(
     response_model_exclude_none=True
 )
 def get_collection_images(
-    id: PyObjectId,
+    id: PydanticObjectId,
     pagination: Pagination,
     user: User | None = Depends(get_optional_user)
 ):
@@ -171,7 +172,7 @@ def get_collection_images(
         }
     image_dicts = list(db_images.find(filters).sort("_id", DESCENDING).limit(pagination.limit))
     if user:
-        for i, image in enumerate(map(lambda image_dict: Image.parse_obj(image_dict), image_dicts)):
+        for i, image in enumerate(map(lambda image_dict: Image.model_validate(image_dict), image_dicts)):
             if image.is_private and user and not compare_digest(image.owner, user.username):
                 del image_dicts[i]
     return image_dicts
@@ -181,7 +182,7 @@ def get_collection_images(
     response_model=list[str]
 )
 def get_collection_images_query_suggestions(
-    id: PyObjectId,
+    id: PydanticObjectId,
     query: str = Body(...),
     user: User | None = Depends(get_optional_user)
 ):
@@ -197,7 +198,7 @@ def get_collection_images_query_suggestions(
         filters["is_private"] = False
     image_dicts = list(db_images.find(filters).sort("_id", DESCENDING).limit(6))
     if user:
-        for i, image in enumerate(map(lambda image_dict: Image.parse_obj(image_dict), image_dicts)):
+        for i, image in enumerate(map(lambda image_dict: Image.model_validate(image_dict), image_dicts)):
             if image.is_private and user and not compare_digest(image.owner, user.username):
                 del image_dicts[i]
     return list(