RED: Add schema, fail on purpose to check

dev.py

@@ -6,7 +6,7 @@ import uvicorn
 import database
 import models
 import crud
-
+import schemas
 
 app = FastAPI()
 
@@ -25,6 +25,23 @@ async def root():
     return {"message": "Hello World"}
 
 
+@app.post("/pydantic_movies/")
+async def create_movie(payload: schemas.MoviePayload, db: Session = Depends(get_db)):
+    data = payload.dict()
+    crud_params = dict(
+        genres=data.get("genres", ["Unknown"]),
+        description=data.get("description", ""),
+        title=data.get("title", ""),
+        vote_average=data.get("vote_average"),
+        vote_count=data.get("vote_count"),
+    )
+
+    movie = crud.create_movie(db, **crud_params)
+    out = {"message": f"Created {movie.title} XX", "id": movie.id}
+    raise ValueError()
+    return out
+
+
 @app.post("/movies/")
 async def create_movie(
     name: str = "", db: Session = Depends(get_db), request: Request = None

schemas.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+
+class MoviePayload(BaseModel):
+    title: str
+
+    vote_count: int
+    vote_average: float
+
+    genres: list[str]
+    description: str
+    release_date: str

utests/test_api.py

@@ -125,6 +125,45 @@ class ApiTestCase(unittest.TestCase):
                         == be_the_fun_in_de_funes[attribute_name]
                     )
 
+    def test_payload_content_in_and_out_loopback_pydantic(self):
+        be_the_fun_in_de_funes = {
+            "id": 1,
+            "title": "La Grande Vadrouille",
+            "description": "During World War II, two French civilians and a downed English Bomber Crew set "
+            "out from Paris to cross the demarcation line between Nazi-occupied Northern France and the "
+            "South. From there they will be able to escape to England. First, they must avoid German troops -"
+            "and the consequences of their own blunders.",
+            "genres": ["Comedy", "War"],
+            "release_date": "1966-12-07",
+            "vote_average": 7.7,
+            "vote_count": 1123,
+        }
+
+        domain_keys = sorted(
+            {k for k in be_the_fun_in_de_funes if k not in ["id"]}
+        )  # Make it deterministic
+        non_primtive = ["genres", "release_date"]
+
+        payload = {k: be_the_fun_in_de_funes[k] for k in domain_keys}
+        # FIXME
+        response = client.post("/pydantic_movies/", json=payload)
+
+        assert response.status_code == 200
+        movie_id = response.json()["id"]
+
+        loopback_fetch = client.get(f"/movies/{movie_id}")
+        assert loopback_fetch.status_code == 200
+        loopback_payload = loopback_fetch.json()
+        # check for keys
+        for attribute_name in domain_keys:
+            with self.subTest(attribute_name=attribute_name):
+                assert attribute_name in loopback_payload
+                if attribute_name not in non_primtive:
+                    assert (
+                        loopback_payload[attribute_name]
+                        == be_the_fun_in_de_funes[attribute_name]
+                    )
+
     @unittest.expectedFailure
     def test_payload_content_in_and_out_loopback_non_primitive(self):
         be_the_fun_in_de_funes = {