More compat

Author: patrick91

src/fastapi_cloud_cli/commands/deploy.py

@@ -13,7 +13,7 @@
 import rignore
 import typer
 from httpx import Client
-from pydantic import BaseModel, EmailStr, TypeAdapter, ValidationError
+from pydantic import BaseModel, EmailStr, ValidationError
 from rich.text import Text
 from rich_toolkit import RichToolkit
 from rich_toolkit.menu import Option
@@ -24,7 +24,11 @@
 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
+from fastapi_cloud_cli.utils.pydantic_compat import (
+    TypeAdapter,
+    model_dump,
+    model_validate,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -417,9 +421,7 @@ def _send_waitlist_form(
     with toolkit.progress("Sending your request...") as progress:
         with APIClient() as client:
             with handle_http_errors(progress):
-                response = client.post(
-                    "/users/waiting-list", json=result.model_dump(mode="json")
-                )
+                response = client.post("/users/waiting-list", json=model_dump(result))
 
                 response.raise_for_status()
 
@@ -444,7 +446,7 @@ def _waitlist_form(toolkit: RichToolkit) -> None:
 
     toolkit.print_line()
 
-    result = SignupToWaitingList(email=email)
+    result = model_validate(SignupToWaitingList, {"email": email})
 
     if toolkit.confirm(
         "Do you want to get access faster by giving us more information?",

src/fastapi_cloud_cli/commands/login.py

@@ -16,7 +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
+from fastapi_cloud_cli.utils.pydantic_compat import model_validate_json
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +44,7 @@ def _start_device_authorization(
 
     response.raise_for_status()
 
-    return model_validate(AuthorizationData, response.json())
+    return model_validate_json(AuthorizationData, response.text)
 
 
 def _fetch_access_token(client: httpx.Client, device_code: str, interval: int) -> str:
@@ -74,7 +74,7 @@ def _fetch_access_token(client: httpx.Client, device_code: str, interval: int) -
 
         time.sleep(interval)
 
-    response_data = model_validate(TokenResponse, response.json())
+    response_data = model_validate_json(TokenResponse, response.text)
 
     return response_data.access_token
 

src/fastapi_cloud_cli/utils/apps.py

@@ -4,7 +4,7 @@
 
 from pydantic import BaseModel
 
-from fastapi_cloud_cli.utils.pydantic_compat import model_validate_json
+from fastapi_cloud_cli.utils.pydantic_compat import model_dump_json, model_validate_json
 
 logger = logging.getLogger("fastapi_cli")
 
@@ -52,7 +52,7 @@ def write_app_config(path_to_deploy: Path, app_config: AppConfig) -> None:
     config_path.parent.mkdir(parents=True, exist_ok=True)
 
     config_path.write_text(
-        app_config.model_dump_json(),
+        model_dump_json(app_config),
         encoding="utf-8",
     )
     readme_path.write_text(README, encoding="utf-8")

src/fastapi_cloud_cli/utils/auth.py

@@ -7,7 +7,7 @@
 
 from pydantic import BaseModel
 
-from fastapi_cloud_cli.utils.pydantic_compat import model_validate_json
+from fastapi_cloud_cli.utils.pydantic_compat import model_dump_json, model_validate_json
 
 from .config import get_auth_path
 
@@ -22,7 +22,7 @@ def write_auth_config(auth_data: AuthConfig) -> None:
     auth_path = get_auth_path()
     logger.debug("Writing auth config to: %s", auth_path)
 
-    auth_path.write_text(auth_data.model_dump_json(), encoding="utf-8")
+    auth_path.write_text(model_dump_json(auth_data), encoding="utf-8")
     logger.debug("Auth config written successfully")
 
 

src/fastapi_cloud_cli/utils/pydantic_compat.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Type, TypeVar
+from typing import Any, Dict, Generic, Type, TypeVar
 
 from pydantic import BaseModel
 from pydantic.version import VERSION as PYDANTIC_VERSION
@@ -7,6 +7,7 @@
 PYDANTIC_V2 = PYDANTIC_VERSION_MINOR_TUPLE[0] == 2
 
 
+T = TypeVar("T")
 Model = TypeVar("Model", bound=BaseModel)
 
 
@@ -22,3 +23,40 @@ def model_validate_json(model_class: Type[Model], data: str) -> Model:
         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]
+
+
+def model_dump(obj: BaseModel, **kwargs: Any) -> Dict[Any, Any]:
+    if PYDANTIC_V2:
+        return obj.model_dump(**kwargs)  # type: ignore[no-any-return, unused-ignore, attr-defined]
+    else:
+        return obj.dict(**kwargs)  # type: ignore[no-any-return, unused-ignore, attr-defined]
+
+
+def model_dump_json(obj: BaseModel) -> str:
+    if PYDANTIC_V2:
+        return obj.model_dump_json()  # type: ignore[no-any-return, unused-ignore, attr-defined]
+    else:
+        return obj.json()  # type: ignore[no-any-return, unused-ignore, attr-defined]
+
+
+class TypeAdapter(Generic[T]):
+    def __init__(self, type_: Type[T]) -> None:
+        self.type_ = type_
+
+        if PYDANTIC_V2:
+            from pydantic import (  # type: ignore[attr-defined, unused-ignore]
+                TypeAdapter as PydanticTypeAdapter,
+            )
+
+            self._adapter = PydanticTypeAdapter(type_)
+        else:
+            self._adapter = None  # type: ignore[assignment, unused-ignore]
+
+    def validate_python(self, value: Any) -> T:
+        """Validate a Python object against the type."""
+        if PYDANTIC_V2:
+            return self._adapter.validate_python(value)  # type: ignore[no-any-return, union-attr, unused-ignore]
+        else:
+            from pydantic import parse_obj_as
+
+            return parse_obj_as(self.type_, value)  # type: ignore[no-any-return, unused-ignore]