Add pydantic compat

Author: patrick91

src/fastapi_cloud_cli/commands/deploy.py

@@ -24,6 +24,7 @@
 from fastapi_cloud_cli.utils.apps import AppConfig, get_app_config, write_app_config
 from fastapi_cloud_cli.utils.auth import is_logged_in
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
+from fastapi_cloud_cli.utils.pydantic_compat import model_validate
 
 logger = logging.getLogger(__name__)
 
@@ -91,7 +92,7 @@ def _get_teams() -> List[Team]:
 
         data = response.json()["data"]
 
-    return [Team.model_validate(team) for team in data]
+    return [model_validate(Team, team) for team in data]
 
 
 class AppResponse(BaseModel):
@@ -108,7 +109,7 @@ def _create_app(team_id: str, app_name: str) -> AppResponse:
 
         response.raise_for_status()
 
-        return AppResponse.model_validate(response.json())
+        return model_validate(AppResponse, response.json())
 
 
 class DeploymentStatus(str, Enum):
@@ -161,7 +162,7 @@ def _create_deployment(app_id: str) -> CreateDeploymentResponse:
         response = client.post(f"/apps/{app_id}/deployments/")
         response.raise_for_status()
 
-        return CreateDeploymentResponse.model_validate(response.json())
+        return model_validate(CreateDeploymentResponse, response.json())
 
 
 class RequestUploadResponse(BaseModel):
@@ -186,7 +187,7 @@ def _upload_deployment(deployment_id: str, archive_path: Path) -> None:
         response = fastapi_client.post(f"/deployments/{deployment_id}/upload")
         response.raise_for_status()
 
-        upload_data = RequestUploadResponse.model_validate(response.json())
+        upload_data = model_validate(RequestUploadResponse, response.json())
         logger.debug("Received upload URL: %s", upload_data.url)
 
         # Upload the archive
@@ -221,7 +222,7 @@ def _get_app(app_slug: str) -> Optional[AppResponse]:
 
         data = response.json()
 
-    return AppResponse.model_validate(data)
+    return model_validate(AppResponse, data)
 
 
 def _get_apps(team_id: str) -> List[AppResponse]:
@@ -231,7 +232,7 @@ def _get_apps(team_id: str) -> List[AppResponse]:
 
         data = response.json()["data"]
 
-    return [AppResponse.model_validate(app) for app in data]
+    return [model_validate(AppResponse, app) for app in data]
 
 
 def _stream_build_logs(deployment_id: str) -> Generator[str, None, None]:
@@ -467,11 +468,12 @@ def _waitlist_form(toolkit: RichToolkit) -> None:
         result = form.run()  # type: ignore
 
         try:
-            result = SignupToWaitingList.model_validate(
+            result = model_validate(
+                SignupToWaitingList,
                 {
                     "email": email,
                     **result,  # type: ignore
-                }
+                },
             )
         except ValidationError:
             toolkit.print(

src/fastapi_cloud_cli/commands/env.py

@@ -11,6 +11,7 @@
 from fastapi_cloud_cli.utils.auth import is_logged_in
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
 from fastapi_cloud_cli.utils.env import validate_environment_variable_name
+from fastapi_cloud_cli.utils.pydantic_compat import model_validate
 
 logger = logging.getLogger(__name__)
 
@@ -29,7 +30,7 @@ def _get_environment_variables(app_id: str) -> EnvironmentVariableResponse:
         response = client.get(f"/apps/{app_id}/environment-variables/")
         response.raise_for_status()
 
-        return EnvironmentVariableResponse.model_validate(response.json())
+        return model_validate(EnvironmentVariableResponse, response.json())
 
 
 def _delete_environment_variable(app_id: str, name: str) -> bool:

src/fastapi_cloud_cli/commands/login.py

@@ -16,6 +16,7 @@
     write_auth_config,
 )
 from fastapi_cloud_cli.utils.cli import get_rich_toolkit, handle_http_errors
+from fastapi_cloud_cli.utils.pydantic_compat import model_validate
 
 logger = logging.getLogger(__name__)
 
@@ -43,7 +44,7 @@ def _start_device_authorization(
 
     response.raise_for_status()
 
-    return AuthorizationData.model_validate(response.json())
+    return model_validate(AuthorizationData, response.json())
 
 
 def _fetch_access_token(client: httpx.Client, device_code: str, interval: int) -> str:
@@ -73,7 +74,7 @@ def _fetch_access_token(client: httpx.Client, device_code: str, interval: int) -
 
         time.sleep(interval)
 
-    response_data = TokenResponse.model_validate(response.json())
+    response_data = model_validate(TokenResponse, response.json())
 
     return response_data.access_token
 

src/fastapi_cloud_cli/utils/apps.py

@@ -4,6 +4,8 @@
 
 from pydantic import BaseModel
 
+from fastapi_cloud_cli.utils.pydantic_compat import model_validate_json
+
 logger = logging.getLogger("fastapi_cli")
 
 
@@ -21,7 +23,7 @@ def get_app_config(path_to_deploy: Path) -> Optional[AppConfig]:
         return None
 
     logger.debug("App config loaded successfully")
-    return AppConfig.model_validate_json(config_path.read_text(encoding="utf-8"))
+    return model_validate_json(AppConfig, config_path.read_text(encoding="utf-8"))
 
 
 README = """

src/fastapi_cloud_cli/utils/auth.py

@@ -7,6 +7,8 @@
 
 from pydantic import BaseModel
 
+from fastapi_cloud_cli.utils.pydantic_compat import model_validate_json
+
 from .config import get_auth_path
 
 logger = logging.getLogger("fastapi_cli")
@@ -44,7 +46,7 @@ def read_auth_config() -> Optional[AuthConfig]:
         return None
 
     logger.debug("Auth config loaded successfully")
-    return AuthConfig.model_validate_json(auth_path.read_text(encoding="utf-8"))
+    return model_validate_json(AuthConfig, auth_path.read_text(encoding="utf-8"))
 
 
 def get_auth_token() -> Optional[str]:

src/fastapi_cloud_cli/utils/pydantic_compat.py

@@ -0,0 +1,24 @@
+from typing import Any, Dict, Type, TypeVar
+
+from pydantic import BaseModel
+from pydantic.version import VERSION as PYDANTIC_VERSION
+
+PYDANTIC_VERSION_MINOR_TUPLE = tuple(int(x) for x in PYDANTIC_VERSION.split(".")[:2])
+PYDANTIC_V2 = PYDANTIC_VERSION_MINOR_TUPLE[0] == 2
+
+
+Model = TypeVar("Model", bound=BaseModel)
+
+
+def model_validate(model_class: Type[Model], data: Dict[Any, Any]) -> Model:
+    if PYDANTIC_V2:
+        return model_class.model_validate(data)  # type: ignore[no-any-return, unused-ignore, attr-defined]
+    else:
+        return model_class.parse_obj(data)  # type: ignore[no-any-return, unused-ignore, attr-defined]
+
+
+def model_validate_json(model_class: Type[Model], data: str) -> Model:
+    if PYDANTIC_V2:
+        return model_class.model_validate_json(data)  # type: ignore[no-any-return, unused-ignore, attr-defined]
+    else:
+        return model_class.parse_raw(data)  # type: ignore[no-any-return, unused-ignore, attr-defined]