Merge pull request #8 from gmos/fix-pydantic

Fixed Pydantic issues and improved Pydantic related tests

Author: bibektimilsina00

.coverage

@@ -2,3 +2,5 @@ __pycache__
 .vscode
 env
 app.egg-info
+.coverage
+.pytest_cache

.gitignore

@@ -2,16 +2,14 @@ FROM python:3.11-slim
 
 WORKDIR /app
 ENV PYTHONPATH=/app
+ENV PYTHONDONTWRITEBYTECODE=1
 
-
-COPY . /app
-
+# First copy only requirements.txt to cache dependencies independently
+COPY requirements.txt /app
 RUN pip install --no-cache-dir -r requirements.txt
 
+COPY . /app
 
 EXPOSE 80
 
 ENV UVICORN_HOST=0.0.0.0 UVICORN_PORT=80 UVICORN_LOG_LEVEL=info
-
-
-

Dockerfile

@@ -1,11 +1,10 @@
 import secrets
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, List, Optional, Union
 
-from pydantic import AnyHttpUrl, EmailStr, HttpUrl, PostgresDsn, validator
-from pydantic_settings import BaseSettings
 from dotenv import load_dotenv
-
-load_dotenv()
+from pydantic import (AnyHttpUrl, AnyUrl, EmailStr, HttpUrl, PostgresDsn,
+                      ValidationInfo, field_validator)
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
 class Settings(BaseSettings):
@@ -20,7 +19,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: Union[str, List[str]]) -> Union[List[str], str]:
         if isinstance(v, str) and not v.startswith("["):
             return [i.strip() for i in v.split(",")]
@@ -31,7 +31,8 @@ def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str
     PROJECT_NAME: str
     SENTRY_DSN: Optional[HttpUrl] = None
 
-    @validator("SENTRY_DSN", pre=True)
+    @field_validator("SENTRY_DSN", mode="before")
+    @classmethod
     def sentry_dsn_can_be_blank(cls, v: Optional[str]) -> Optional[str]:
         if v is None or len(v) == 0:
             return None
@@ -43,14 +44,15 @@ def sentry_dsn_can_be_blank(cls, v: Optional[str]) -> Optional[str]:
     POSTGRES_DB: str
     SQLALCHEMY_DATABASE_URI: Optional[PostgresDsn] = None
 
-    @validator("SQLALCHEMY_DATABASE_URI", pre=True)
-    def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any:
+    @field_validator("SQLALCHEMY_DATABASE_URI", mode="before")
+    @classmethod
+    def assemble_db_connection(cls, v: Optional[str], info: ValidationInfo) -> Any:
         if isinstance(v, str):
             return v
-        user = values.get("POSTGRES_USER")
-        password = values.get("POSTGRES_PASSWORD")
-        host = values.get("POSTGRES_SERVER")
-        db = values.get("POSTGRES_DB")
+        user = info.data.get("POSTGRES_USER")
+        password = info.data.get("POSTGRES_PASSWORD")
+        host = info.data.get("POSTGRES_SERVER")
+        db = info.data.get("POSTGRES_DB")
 
         if all([user, password, host, db]):
             return f"postgresql://{user}:{password}@{host}/{db}"
@@ -65,31 +67,32 @@ def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]) -> Any
     EMAILS_FROM_EMAIL: Optional[EmailStr] = None
     EMAILS_FROM_NAME: Optional[str] = None
 
-    @validator("EMAILS_FROM_NAME")
-    def get_project_name(cls, v: Optional[str], values: Dict[str, Any]) -> str:
+    @field_validator("EMAILS_FROM_NAME", mode="before")
+    @classmethod
+    def get_project_name(cls, v: Optional[str], info: ValidationInfo) -> str:
         if not v:
-            return values["PROJECT_NAME"]
+            return info.data.get("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")
+    @classmethod
+    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
+    EMAIL_TEST_USER: EmailStr = "test@example.com"
     FIRST_SUPERUSER: EmailStr
     FIRST_SUPERUSER_PASSWORD: str
     USERS_OPEN_REGISTRATION: bool = False
-
-    class Config:
-        case_sensitive = True
+    model_config = SettingsConfigDict(case_sensitive=True)
 
 
+load_dotenv()
 settings = Settings()

app/core/config.py

@@ -50,7 +50,7 @@ def update(
         if isinstance(obj_in, dict):
             update_data = obj_in
         else:
-            update_data = obj_in.dict(exclude_unset=True)
+            update_data = obj_in.model_dump(exclude_unset=True)
         for field in obj_data:
             if field in update_data:
                 setattr(db_obj, field, update_data[field])

app/crud/base.py

@@ -30,7 +30,7 @@ def update(
         if isinstance(obj_in, dict):
             update_data = obj_in
         else:
-            update_data = obj_in.dict(exclude_unset=True)
+            update_data = obj_in.model_dump(exclude_unset=True)
         if update_data["password"]:
             hashed_password = get_password_hash(update_data["password"])
             del update_data["password"]

app/crud/crud_user.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 
 
 # Shared properties
@@ -24,9 +24,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

app/schemas/item.py

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

app/schemas/user.py

@@ -0,0 +1,76 @@
+from io import StringIO
+from typing import Any
+import os
+
+from dotenv import load_dotenv
+
+from app.core.config import Settings
+from app.tests.utils.utils import random_email, random_lower_string, random_url
+
+
+def make_settings(env_items: dict[str, Any]):
+    os.environ.clear()
+    env_file_settings = StringIO()
+    for key, value in env_items.items():
+        env_file_settings.write(f"{key}={value}\n")
+    env_file_settings.seek(0)
+    load_dotenv(stream=env_file_settings)
+    return Settings()
+
+
+MANDATORY = {
+    "FIRST_SUPERUSER_PASSWORD": random_lower_string(),
+    "FIRST_SUPERUSER": random_email(),
+    "POSTGRES_DB": random_lower_string(),
+    "POSTGRES_PASSWORD": random_lower_string(),
+    "POSTGRES_SERVER": random_lower_string(),
+    "POSTGRES_USER": random_lower_string(),
+    "PROJECT_NAME": random_lower_string(),
+    "SERVER_HOST": random_url(),
+    "SERVER_NAME": random_lower_string(),
+}
+
+
+def test_mandatory_and_defaults() -> None:
+    settings = make_settings(MANDATORY)
+    for key, value in MANDATORY.items():
+        assert str(getattr(settings, key)) == str(value)
+    assert settings.EMAIL_TEMPLATES_DIR == "/app/app/email-templates/build"
+    assert settings.EMAILS_ENABLED is False
+    assert settings.EMAILS_FROM_EMAIL is None
+    assert settings.EMAILS_FROM_NAME == settings.PROJECT_NAME
+    assert settings.EMAIL_RESET_TOKEN_EXPIRE_HOURS == 48
+    assert settings.EMAIL_TEST_USER == "test@example.com"
+    assert settings.EMAIL_TEMPLATES_DIR == "/app/app/email-templates/build"
+
+
+def test_assemble_db_connection() -> None:
+    settings = make_settings(MANDATORY)
+    assert str(settings.SQLALCHEMY_DATABASE_URI) == (
+        f"postgresql://{settings.POSTGRES_USER}:"
+        f"{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_SERVER}/"
+        f"{settings.POSTGRES_DB}"
+    )
+
+
+def test_backend_cors_origins() -> None:
+    settings = make_settings(
+        MANDATORY
+        | {"BACKEND_CORS_ORIGINS": '["http://localhost", "http://localhost:3000"]'}
+    )
+    assert [str(item) for item in settings.BACKEND_CORS_ORIGINS] == [
+        "http://localhost/",
+        "http://localhost:3000/",
+    ]
+
+
+def test_email_enabled() -> None:
+    settings = make_settings(
+        MANDATORY
+        | {
+            "SMTP_HOST": "www.example.com",
+            "SMTP_PORT": 25,
+            "EMAILS_FROM_EMAIL": random_email(),
+        }
+    )
+    assert settings.EMAILS_ENABLED is True

app/tests/settings/test_settings.py

@@ -15,6 +15,10 @@ def random_email() -> str:
     return f"{random_lower_string()}@{random_lower_string()}.com"
 
 
+def random_url() -> str:
+    return f"https://{random_lower_string()}.com/"
+
+
 def get_superuser_token_headers(client: TestClient) -> Dict[str, str]:
     login_data = {
         "username": settings.FIRST_SUPERUSER,