Make user_id mandatory

Author: pentest-gg

fastapi_users_db_dynamodb/access_token.py

@@ -161,6 +161,9 @@ async def create(self, create_dict: dict[str, Any] | AP) -> AP:
             token = self.access_token_table(**create_dict)
         else:
             token = create_dict
+        if not getattr(token, "user_id", None):
+            raise ValueError("AccessToken must implement and store value 'user_id'.")
+
         try:
             await token.save(condition=self.access_token_table.token.does_not_exist())  # type: ignore
         except PutError as e:

tests/test_access_token.py

@@ -126,6 +126,12 @@ async def test_queries(
     access_token = await dynamodb_access_token_db.create(access_token_create)
     assert access_token.token == "TOKEN"
     assert access_token.user_id == user_id
+    with pytest.raises(
+        ValueError,
+        match="AccessToken must implement and store value 'user_id'.",
+    ):
+        access_token_create.pop("user_id")
+        await dynamodb_access_token_db.create(access_token_create)
 
     # Update
     new_time = now_utc()