migrate to new verstion of fastAPI with large breaking changes in Pyd…

…antic

requirements.dev.txt

@@ -2,4 +2,4 @@ flake8==6.0.0
 pytest==7.2.1
 httpx==0.23.3
 coverage==7.2.3
-polyfactory==2.0.0
+polyfactory==2.7.2

requirements.txt

@@ -1,11 +1,12 @@
 redis==4.4.4
-fastapi==0.97.0
-uvicorn[standard]==0.20.0
+fastapi==0.101.0
+uvicorn[standard]==0.23.2
 pyjwt[crypto]==2.6.0
-fastapi_mail==1.2.5
 cfenv==0.5.3
 SQLAlchemy==2.0.5.post1
 psycopg2==2.9.5
 alembic==1.10.2
 PyMuPDF==1.21.1
+pydantic-settings==2.0.2
+email-validator==2.0.0.post2
 python-multipart==0.0.6

training/api/api_v1/auth.py

@@ -34,15 +34,15 @@ def auth_exchange(
             detail="Invalid user."
         )
 
-    user = User.from_orm(db_user)
+    user = User.model_validate(db_user)
     if not user.is_admin():
         logging.info(f"UAA authenticated, but not an admin: {uaa_user['email']}")
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="Not authorized to login."
         )
 
-    jwt_user = UserJWT.from_orm(db_user)
-    encoded_jwt = jwt.encode(jwt_user.dict(), settings.JWT_SECRET, algorithm="HS256")
+    jwt_user = UserJWT.model_validate(db_user)
+    encoded_jwt = jwt.encode(jwt_user.model_dump(), settings.JWT_SECRET, algorithm="HS256")
     logging.info(f"Token exchange success for {db_user.email}")
     return {'user': jwt_user, 'jwt': encoded_jwt}

training/api/api_v1/loginless_flow.py

@@ -2,7 +2,7 @@
 import jwt
 from typing import Union
 
-from fastapi import APIRouter, status, Response, HTTPException, Depends
+from fastapi import APIRouter, status, Response, HTTPException, Depends, Request
 from training.schemas import TempUser, IncompleteTempUser, WebDestination, UserJWT
 from training.data import UserCache
 from training.repositories import UserRepository
@@ -32,14 +32,18 @@ def page_lookup():
 
 
 @router.post("/get-link", status_code=status.HTTP_201_CREATED)
-def send_link(
+async def send_link(
     response: Response,
+    request: Request,
     user: Union[TempUser, IncompleteTempUser],
     dest: WebDestination,
     repo: UserRepository = Depends(user_repository),
     cache: UserCache = Depends(UserCache),
     page_id_lookup: dict = Depends(page_lookup)
 ):
+    tu = await request.json()
+    print("res: ", tu)
+    print("user: ", type(user))
     try:
         required_roles = page_id_lookup[dest.page_id]['required_roles']
     except KeyError:
@@ -63,7 +67,7 @@ def send_link(
                     detail="Unauthorized"
                 )
 
-            user = TempUser.parse_obj({
+            user = TempUser.model_validate({
                 "name": user_from_db.name,
                 "email": user_from_db.email,
                 "agency_id": user_from_db.agency_id,
@@ -118,7 +122,7 @@ async def get_user(
     db_user = repo.find_by_email(user.email)
     if not db_user:
         db_user = repo.create(user)
-    user_return = UserJWT.from_orm(db_user)
+    user_return = UserJWT.model_validate(db_user)
     logging.info(f"Confirmed email token for {user.email}")
-    encoded_jwt = jwt.encode(user_return.dict(), settings.JWT_SECRET, algorithm="HS256")
+    encoded_jwt = jwt.encode(user_return.model_dump(), settings.JWT_SECRET, algorithm="HS256")
     return {'user': user_return, 'jwt': encoded_jwt}

training/config.py

@@ -1,6 +1,8 @@
-from pydantic import BaseSettings, EmailStr
+from typing import Tuple, Type
+from pydantic import EmailStr
 from typing import Dict, Any
 from cfenv import AppEnv
+from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict
 
 
 def vcap_services_settings(settings: BaseSettings) -> Dict[str, Any]:
@@ -55,7 +57,10 @@ class Settings(BaseSettings):
     SMTP_USER: str | None
     SMTP_SERVER: str
     SMTP_PORT: int
-    EMAIL_FROM: EmailStr = EmailStr("[email protected]")
+    SMTP_STARTTLS: bool
+    SMTP_SSL_TLS: bool
+
+    EMAIL_FROM: EmailStr = "[email protected]"
     EMAIL_FROM_NAME: str = "GSA SmartPay"
     EMAIL_SUBJECT: str = "GSA SmartPay Training"
 
@@ -74,18 +79,26 @@ class Settings(BaseSettings):
     AUTH_CLIENT_ID: str
     AUTH_AUTHORITY_URL: str
 
-    class Config:
-        env_file = '.env'
-        env_file_encoding = 'utf-8'
-
-        @classmethod
-        def customise_sources(cls, init_settings, env_settings, file_secret_settings):
-            return (
-                init_settings,
-                env_settings,
-                file_secret_settings,
-                vcap_services_settings,
-            )
+    model_config = SettingsConfigDict(
+        env_file='.env',
+        env_file_encoding='utf-8'
+    )
+
+    @classmethod
+    def customise_sources(
+        cls,
+        settings_cls: Type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource
+    ) -> Tuple[PydanticBaseSettingsSource, ...]:
+        return (
+            init_settings,
+            env_settings,
+            file_secret_settings,
+            vcap_services_settings,
+        )
 
 
 settings = Settings()  # type: ignore

training/data/user_cache.py

@@ -32,7 +32,7 @@ def get(self, token: str) -> Optional[UserCreate]:
 
     def set(self, user: TempUser) -> str:
         token = str(uuid4())
-        user_str = json.dumps(user.dict())
+        user_str = json.dumps(user.model_dump())
         # try/except here
         redis.set(token, user_str)
         redis.expire(token, self.CACHE_TTL)

training/repositories/quiz.py

@@ -9,7 +9,7 @@ def __init__(self, session: Session):
         super().__init__(session, models.Quiz)
 
     def create(self, quiz: schemas.QuizCreate) -> models.Quiz:
-        content_dict = quiz.content.dict()
+        content_dict = quiz.content.model_dump()
 
         # Assign IDs to questions and choices
         for qindex, question in enumerate(content_dict.get("questions", [])):

training/schemas/agency.py

@@ -1,10 +1,10 @@
-from pydantic import BaseModel
-from pydantic.schema import Optional
+from typing import Optional
+from pydantic import ConfigDict, BaseModel
 
 
 class AgencyBase(BaseModel):
     name: str
-    bureau: Optional[str]
+    bureau: Optional[str] = None
 
 
 class AgencyCreate(AgencyBase):
@@ -13,9 +13,7 @@ class AgencyCreate(AgencyBase):
 
 class Agency(AgencyBase):
     id: int
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)
 
 
 class Bureau(BaseModel):

training/schemas/quiz.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 from training.schemas import QuizContent, QuizContentCreate, QuizContentPublic
 
 
@@ -29,13 +29,9 @@ class QuizCreate(QuizBase):
 class QuizPublic(QuizBase):
     id: int
     content: QuizContentPublic
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)
 
 
 class Quiz(QuizBase):
     id: int
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)

training/schemas/quiz_choice.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 
 
 class QuizChoiceBase(BaseModel):
@@ -16,6 +16,4 @@ class QuizChoicePublic(QuizChoiceBase):
 class QuizChoice(QuizChoiceBase):
     id: int
     correct: bool
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)

training/schemas/quiz_completion.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 
 
 class QuizCompletionBase(BaseModel):
@@ -15,6 +15,4 @@ class QuizCompletionCreate(QuizCompletionBase):
 class QuizCompletion(QuizCompletionBase):
     id: int
     submit_ts: datetime
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)

training/schemas/quiz_grade.py

@@ -9,7 +9,7 @@ class QuizGradeQuestion(BaseModel):
 
 
 class QuizGrade(BaseModel):
-    quiz_completion_id: int | None
+    quiz_completion_id: int | None = None
     quiz_id: int
     correct_count: int
     question_count: int

training/schemas/quiz_question.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 from training.schemas import QuizChoice, QuizChoicePublic, QuizChoiceCreate
 
 
@@ -25,6 +25,4 @@ class QuizQuestionPublic(QuizQuestionBase):
 
 class QuizQuestion(QuizQuestionBase):
     id: int
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)

training/schemas/report_user_x_agency.py

@@ -1,9 +1,7 @@
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 
 
 class ReportUserXAgency(BaseModel):
     user_id: int
     agency_id: int
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)

training/schemas/role.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import ConfigDict, BaseModel
 
 
 class RoleCreate(BaseModel):
@@ -7,6 +7,4 @@ class RoleCreate(BaseModel):
 
 class Role(RoleCreate):
     id: int
-
-    class Config:
-        orm_mode = True
+    model_config = ConfigDict(from_attributes=True)

training/schemas/temp_user.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, EmailStr
+from pydantic import ConfigDict, BaseModel, EmailStr, field_validator
 
 
 class TempUser(BaseModel):
@@ -9,9 +9,21 @@ class TempUser(BaseModel):
     email: EmailStr
     name: str
     agency_id: int
+    model_config = ConfigDict(from_attributes=True)
 
-    class Config:
-        orm_mode = True
+    @field_validator("agency_id", mode="before")
+    @classmethod
+    def to_int(cls, value: str | int) -> int:
+        '''
+            This addresses a bug in pydantic that does not correctly choose
+            the right value from the union[TempUser, IncompleteTempUser]
+            when agency_id is a string. Related:
+            https://docs.pydantic.dev/dev-v2/migration/#unions
+        '''
+        if isinstance(value, str):
+            value = int(value)
+
+        return value
 
 
 class IncompleteTempUser(BaseModel):