⬆ Upgrade code to support pydantic V2 (#615)

src/backend/app/api/api_v1/endpoints/users.py

@@ -1,7 +1,7 @@
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
-from sqlmodel import func, select, delete
+from sqlmodel import delete, func, select
 
 from app import crud
 from app.api.deps import (
@@ -12,6 +12,7 @@
 from app.core.config import settings
 from app.core.security import get_password_hash, verify_password
 from app.models import (
+    Item,
     Message,
     UpdatePassword,
     User,
@@ -21,7 +22,6 @@
     UsersOut,
     UserUpdate,
     UserUpdateMe,
-    Item
 )
 from app.utils import send_new_account_email
 

src/backend/app/core/config.py

@@ -1,7 +1,14 @@
 import secrets
 from typing import Any
 
-from pydantic import AnyHttpUrl, BaseSettings, EmailStr, HttpUrl, PostgresDsn, validator
+from pydantic import (
+    AnyHttpUrl,
+    HttpUrl,
+    PostgresDsn,
+    ValidationInfo,
+    field_validator,
+)
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
 class Settings(BaseSettings):
@@ -16,7 +23,8 @@ class Settings(BaseSettings):
     # "http://localhost:8080", "http://local.dockertoolbox.tiangolo.com"]'
     BACKEND_CORS_ORIGINS: list[AnyHttpUrl] = []
 
-    @validator("BACKEND_CORS_ORIGINS", pre=True)
+    @field_validator("BACKEND_CORS_ORIGINS", mode="before")
+    @classmethod
     def assemble_cors_origins(cls, v: str | list[str]) -> list[str] | str:
         if isinstance(v, str) and not v.startswith("["):
             return [i.strip() for i in v.split(",")]
@@ -27,7 +35,8 @@ def assemble_cors_origins(cls, v: str | list[str]) -> list[str] | str:
     PROJECT_NAME: str
     SENTRY_DSN: HttpUrl | None = None
 
-    @validator("SENTRY_DSN", pre=True)
+    @field_validator("SENTRY_DSN", mode="before")
+    @classmethod
     def sentry_dsn_can_be_blank(cls, v: str) -> str | None:
         if len(v) == 0:
             return None
@@ -39,51 +48,49 @@ def sentry_dsn_can_be_blank(cls, v: str) -> str | None:
     POSTGRES_DB: str
     SQLALCHEMY_DATABASE_URI: PostgresDsn | None = None
 
-    @validator("SQLALCHEMY_DATABASE_URI", pre=True)
-    def assemble_db_connection(cls, v: str | None, values: dict[str, Any]) -> Any:
+    @field_validator("SQLALCHEMY_DATABASE_URI", mode="before")
+    def assemble_db_connection(cls, v: str | None, info: ValidationInfo) -> Any:
         if isinstance(v, str):
             return v
         return PostgresDsn.build(
             scheme="postgresql+psycopg",
-            user=values.get("POSTGRES_USER"),
-            password=values.get("POSTGRES_PASSWORD"),
-            host=values.get("POSTGRES_SERVER"),
-            path=f"/{values.get('POSTGRES_DB') or ''}",
+            username=info.data.get("POSTGRES_USER"),
+            password=info.data.get("POSTGRES_PASSWORD"),
+            host=info.data.get("POSTGRES_SERVER"),
+            path=f"{info.data.get('POSTGRES_DB') or ''}",
         )
 
     SMTP_TLS: bool = True
     SMTP_PORT: int | None = None
     SMTP_HOST: str | None = None
     SMTP_USER: str | None = None
     SMTP_PASSWORD: str | None = None
-    EMAILS_FROM_EMAIL: EmailStr | None = None
+    EMAILS_FROM_EMAIL: str | None = None  #TODO: update type to EmailStr when sqlmodel supports it
     EMAILS_FROM_NAME: str | None = None
 
-    @validator("EMAILS_FROM_NAME")
-    def get_project_name(cls, v: str | None, values: dict[str, Any]) -> str:
+    @field_validator("EMAILS_FROM_NAME")
+    def get_project_name(cls, v: str | None, info: ValidationInfo) -> str:
         if not v:
-            return values["PROJECT_NAME"]
+            return info.data["PROJECT_NAME"]
         return v
 
     EMAIL_RESET_TOKEN_EXPIRE_HOURS: int = 48
     EMAIL_TEMPLATES_DIR: str = "/app/app/email-templates/build"
     EMAILS_ENABLED: bool = False
 
-    @validator("EMAILS_ENABLED", pre=True)
-    def get_emails_enabled(cls, v: bool, values: dict[str, Any]) -> bool:
+    @field_validator("EMAILS_ENABLED", mode="before")
+    def get_emails_enabled(cls, v: bool, info: ValidationInfo) -> bool:
         return bool(
-            values.get("SMTP_HOST")
-            and values.get("SMTP_PORT")
-            and values.get("EMAILS_FROM_EMAIL")
+            info.data.get("SMTP_HOST")
+            and info.data.get("SMTP_PORT")
+            and info.data.get("EMAILS_FROM_EMAIL")
         )
 
-    EMAIL_TEST_USER: EmailStr = "test@example.com"  # type: ignore
-    FIRST_SUPERUSER: EmailStr
+    EMAIL_TEST_USER: str = "test@example.com"  #TODO: update type to EmailStr when sqlmodel supports it
+    FIRST_SUPERUSER: str  #TODO: update type to EmailStr when sqlmodel supports it
     FIRST_SUPERUSER_PASSWORD: str
     USERS_OPEN_REGISTRATION: bool = False
-
-    class Config:
-        case_sensitive = True
+    model_config = SettingsConfigDict(case_sensitive=True)
 
 
 settings = Settings()

src/backend/app/db/engine.py

@@ -2,4 +2,4 @@
 
 from app.core.config import settings
 
-engine = create_engine(settings.SQLALCHEMY_DATABASE_URI)
+engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))

src/backend/app/models.py

@@ -1,10 +1,10 @@
-from pydantic import EmailStr
 from sqlmodel import Field, Relationship, SQLModel
 
 
 # Shared properties
+# TODO replace email str with EmailStr when sqlmodel supports it
 class UserBase(SQLModel):
-    email: EmailStr = Field(unique=True, index=True)
+    email: str = Field(unique=True, index=True)
     is_active: bool = True
     is_superuser: bool = False
     full_name: str | None = None
@@ -15,21 +15,24 @@ class UserCreate(UserBase):
     password: str
 
 
+# TODO replace email str with EmailStr when sqlmodel supports it
 class UserCreateOpen(SQLModel):
-    email: EmailStr
+    email: str
     password: str
     full_name: str | None = None
 
 
 # Properties to receive via API on update, all are optional
+# TODO replace email str with EmailStr when sqlmodel supports it
 class UserUpdate(UserBase):
-    email: EmailStr | None = None
+    email: str | None = None
     password: str | None = None
 
 
+# TODO replace email str with EmailStr when sqlmodel supports it
 class UserUpdateMe(SQLModel):
     full_name: str | None = None
-    email: EmailStr | None = None
+    email: str | None = None
 
 
 class UpdatePassword(SQLModel):

src/backend/app/schemas/item.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 
 # Shared properties
@@ -22,9 +22,7 @@ class ItemInDBBase(ItemBase):
     id: int
     title: str
     owner_id: int
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)
 
 
 # Properties to return to client

src/backend/app/schemas/user.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, EmailStr
+from pydantic import BaseModel, ConfigDict, EmailStr
 
 
 # Shared properties
@@ -22,9 +22,7 @@ class UserUpdate(UserBase):
 
 class UserInDBBase(UserBase):
     id: int | None = None
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)
 
 
 # Additional properties to return via API

src/backend/pyproject.toml

@@ -13,7 +13,7 @@ email-validator = "^2.1.0.post1"
 celery = "^5.3.5"
 passlib = {extras = ["bcrypt"], version = "^1.7.4"}
 tenacity = "^8.2.3"
-pydantic = "<2.0"
+pydantic = ">2.0"
 emails = "^0.6"
 
 gunicorn = "^21.2.0"
@@ -25,6 +25,7 @@ psycopg = {extras = ["binary"], version = "^3.1.13"}
 sqlmodel = "^0.0.16"
 # Pin bcrypt until passlib supports the latest
 bcrypt = "4.0.1"
+pydantic-settings = "^2.2.1"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.3"