#python #sqlalchemy #fastapi #pydantic
#python #sqlalchemy #fastapi #pydantic
Вопрос:
Я использую fastapi с моделью pydantic и sqlalchemy,чтобы создать api,который выглядит как социальная сеть (создает пользователей,сообщения и голоса) на основе проекта санджеева на YouTube (он недавно был выпущен в лагере свободного кода), и я не знаю, что делать, потому что я скопировал код и почти ничего не изменил, он не может получать данные, которые я ввел в запрос, код приспособления является:
def test_users(client): users_data = [ { "email": "julio@gmail.com", "password": "123" }, { "email": "estelson@gmail.com", "password": "123" }, { "email": "joana@gmail.com", "password": "123" }, { "email": "julie@gmail.com", "password": "123" } ] new_users = list() for user_data in users_data: res = client.post("/users/", json=user_data) assert res.status_code == 201 new_user = res.json() new_user['password'] = user_data['password'] new_users.append(new_user) return new_users @pytest.fixture def test_posts(test_users, session): posts_data = [ { 'title': 'first title', "content": "first content", "owner_id": test_users[0]['id'] }, { "title": "2nd title", "content": "2nd content", "owner_id": test_users[0]['id'] }, { "title": "3rd title", "content": "3rd content", "owner_id": test_users[0]['id'] }, { "title": "3rd title", "content": "3rd content", "owner_id": test_users[1]['id'] }, { "title": "4rd title", "content": "4rd content", "owner_id": test_users[1]['id'] }] def create_post_model(post): return models.Post(**post) post_map = map(create_post_model, posts_data) posts = list(post_map) session.add_all(posts) session.commit() # for post in posts_data: # post_model = models.Post(**post) # session.add(post_model) # session.commit() posts = session.query(models.Post).all() return posts Затем код тестов :
def test_get_all_posts(authorized_client, test_posts): res = authorized_client.get('/posts/') def validate(post): return schemas.PostOut(**post) posts_map = map(validate, list(res.json())) posts_list = list(posts_map) print(posts_list) assert len(res.json()) == len(test_posts) assert res.status_code == 200 def test_get_one_post(authorized_client, test_posts): res = authorized_client.get(f"/posts/{test_posts[0].id}") post = schemas.PostOut(**res.json()) print(post.Post.title, post.Post.content, post.Post.published, post.Post.id) assert post.Post.id == test_posts[0].id assert post.Post.content == test_posts[0].content assert post.Post.title == test_posts[0].title Маршруты:
@router.get('/', response_model=List[schemas.PostOut]) def get_posts(db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user), limit: int = 10, skip: int = 0, search: Optional[str] = ""): posts = db.query(models.Post, func.count(models.Likes.post_id).label('likes'), func.count(models.Dislikes.post_id).label('dislikes')).join( models.Likes, models.Likes.post_id == models.Post.id, isouter=True).join( models.Dislikes, models.Dislikes.post_id == models.Post.id, isouter=True).group_by(models.Post.id).filter(models.Post.title.contains(search)).limit(limit).offset(skip).all() return posts @router.get('/{id}', response_model=schemas.PostOut) def get_post(id: int, db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user)): post = db.query(models.Post, func.count(models.Likes.post_id).label('likes'), func.count(models.Dislikes.post_id).label('dislikes')).join( models.Likes, models.Likes.post_id == models.Post.id, isouter=True).join(models.Dislikes, models.Dislikes.post_id == models.Post.id, isouter=True).group_by(models.Post.id).filter(models.Post.id == id).first() if not post: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'post with id : {id} was not found') return post The schemas :
class PostBase(BaseModel): title: str content: str published: bool = True class Post(PostBase): id: int created_at: datetime owner_id: int owner: UserOut class Config: orm_mode = True class PostOut(PostBase): Post: Post class Config: orm_mode = True And finally the errors
authorized_client = lt;starlette.testclient.TestClient object at 0x00000260702DB5E0gt; test_posts = [lt;app.models.Post object at 0x000002607028C190gt;, lt;app.models.Post object at 0x000002607028C340gt;, lt;app.models.Post object at 0x00000260702DB790gt;, lt;app.models.Post object at 0x000002606FB9BE80gt;, lt;app.models.Post object at 0x000002607027E1C0gt;] def test_get_all_posts(authorized_client, test_posts): gt; res = authorized_client.get('/posts/') teststest_post.py:38: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ venvlibsite-packagesrequestssessions.py:555: in get return self.request('GET', url, **kwargs) venvlibsite-packagesstarlettetestclient.py:468: in request return super().request( venvlibsite-packagesrequestssessions.py:542: in request resp = self.send(prep, **send_kwargs) venvlibsite-packagesrequestssessions.py:655: in send r = adapter.send(request, **kwargs) venvlibsite-packagesstarlettetestclient.py:266: in send raise exc venvlibsite-packagesstarlettetestclient.py:263: in send portal.call(self.app, scope, receive, send) venvlibsite-packagesanyiofrom_thread.py:240: in call return cast(T_Retval, self.start_task_soon(func, *args).result()) C:Python39libconcurrentfutures_base.py:445: in result return self.__get_result() C:Python39libconcurrentfutures_base.py:390: in __get_result raise self._exception venvlibsite-packagesanyiofrom_thread.py:187: in _call_func retval = await retval venvlibsite-packagesfastapiapplications.py:208: in __call__ await super().__call__(scope, receive, send) venvlibsite-packagesstarletteapplications.py:112: in __call__ await self.middleware_stack(scope, receive, send) venvlibsite-packagesstarlettemiddlewareerrors.py:181: in __call__ raise exc venvlibsite-packagesstarlettemiddlewareerrors.py:159: in __call__ await self.app(scope, receive, _send) venvlibsite-packagesstarlettemiddlewarecors.py:84: in __call__ await self.app(scope, receive, send) venvlibsite-packagesstarletteexceptions.py:82: in __call__ raise exc venvlibsite-packagesstarletteexceptions.py:71: in __call__ await self.app(scope, receive, sender) venvlibsite-packagesstarletterouting.py:656: in __call__ await route.handle(scope, receive, send) venvlibsite-packagesstarletterouting.py:259: in handle await self.app(scope, receive, send) venvlibsite-packagesstarletterouting.py:61: in app response = await func(request) venvlibsite-packagesfastapirouting.py:234: in app response_data = await serialize_response( _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ async def serialize_response( *, field: Optional[ModelField] = None, response_content: Any, include: Optional[Union[SetIntStr, DictIntStrAny]] = None, exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None, by_alias: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, is_coroutine: bool = True, ) -gt; Any: if field: errors = [] response_content = _prepare_response_content( response_content, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults, exclude_none=exclude_none, ) if is_coroutine: value, errors_ = field.validate(response_content, {}, loc=("response",)) else: value, errors_ = await run_in_threadpool( field.validate, response_content, {}, loc=("response",) ) if isinstance(errors_, ErrorWrapper): errors.append(errors_) elif isinstance(errors_, list): errors.extend(errors_) if errors: gt; raise ValidationError(errors, field.type_) E pydantic.error_wrappers.ValidationError: 10 validation errors for PostOut E response -gt; 0 -gt; title E field required (type=value_error.missing) E response -gt; 0 -gt; content E field required (type=value_error.missing) E response -gt; 1 -gt; title E field required (type=value_error.missing) E response -gt; 1 -gt; content E field required (type=value_error.missing) E response -gt; 2 -gt; title E field required (type=value_error.missing) E response -gt; 2 -gt; content E field required (type=value_error.missing) E response -gt; 3 -gt; title E field required (type=value_error.missing) E response -gt; 3 -gt; content E field required (type=value_error.missing) E response -gt; 4 -gt; title E field required (type=value_error.missing) E response -gt; 4 -gt; content E field required (type=value_error.missing) venvlibsite-packagesfastapirouting.py:137: ValidationError __________________________________________________________ test_get_one_post ___________________________________________________________ authorized_client = lt;starlette.testclient.TestClient object at 0x0000026070302430gt; test_posts = [lt;app.models.Post object at 0x000002607091D730gt;, lt;app.models.Post object at 0x000002607091D520gt;, lt;app.models.Post object at 0x000002607091DB80gt;, lt;app.models.Post object at 0x000002607091D9D0gt;, lt;app.models.Post object at 0x000002607091D460gt;] def test_get_one_post(authorized_client, test_posts): gt; res = authorized_client.get(f"/posts/{test_posts[0].id}") teststest_post.py:51: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ venvlibsite-packagesrequestssessions.py:555: in get return self.request('GET', url, **kwargs) venvlibsite-packagesstarlettetestclient.py:468: in request return super().request( venvlibsite-packagesrequestssessions.py:542: in request resp = self.send(prep, **send_kwargs) venvlibsite-packagesrequestssessions.py:655: in send r = adapter.send(request, **kwargs) venvlibsite-packagesstarlettetestclient.py:266: in send raise exc venvlibsite-packagesstarlettetestclient.py:263: in send portal.call(self.app, scope, receive, send) venvlibsite-packagesanyiofrom_thread.py:240: in call return cast(T_Retval, self.start_task_soon(func, *args).result()) C:Python39libconcurrentfutures_base.py:445: in result return self.__get_result() C:Python39libconcurrentfutures_base.py:390: in __get_result raise self._exception venvlibsite-packagesanyiofrom_thread.py:187: in _call_func retval = await retval venvlibsite-packagesfastapiapplications.py:208: in __call__ await super().__call__(scope, receive, send) venvlibsite-packagesstarletteapplications.py:112: in __call__ await self.middleware_stack(scope, receive, send) venvlibsite-packagesstarlettemiddlewareerrors.py:181: in __call__ raise exc venvlibsite-packagesstarlettemiddlewareerrors.py:159: in __call__ await self.app(scope, receive, _send) venvlibsite-packagesstarlettemiddlewarecors.py:84: in __call__ await self.app(scope, receive, send) venvlibsite-packagesstarletteexceptions.py:82: in __call__ raise exc venvlibsite-packagesstarletteexceptions.py:71: in __call__ await self.app(scope, receive, sender) venvlibsite-packagesstarletterouting.py:656: in __call__ await route.handle(scope, receive, send) venvlibsite-packagesstarletterouting.py:259: in handle await self.app(scope, receive, send) venvlibsite-packagesstarletterouting.py:61: in app response = await func(request) venvlibsite-packagesfastapirouting.py:234: in app response_data = await serialize_response( _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ async def serialize_response( *, field: Optional[ModelField] = None, response_content: Any, include: Optional[Union[SetIntStr, DictIntStrAny]] = None, exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None, by_alias: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, is_coroutine: bool = True, ) -gt; Any: if field: errors = [] response_content = _prepare_response_content( response_content, exclude_unset=exclude_unset, exclude_defaults=exclude_defaults, exclude_none=exclude_none, ) if is_coroutine: value, errors_ = field.validate(response_content, {}, loc=("response",)) else: value, errors_ = await run_in_threadpool( field.validate, response_content, {}, loc=("response",) ) if isinstance(errors_, ErrorWrapper): errors.append(errors_) elif isinstance(errors_, list): errors.extend(errors_) if errors: gt; raise ValidationError(errors, field.type_) E pydantic.error_wrappers.ValidationError: 2 validation errors for PostOut E response -gt; title E field required (type=value_error.missing) E response -gt; content E field required (type=value_error.missing) venvlibsite-packagesfastapirouting.py:137: ValidationError ======================================================= short test summary info ======================================================== FAILED tests/test_post.py::test_get_all_posts - pydantic.error_wrappers.ValidationError: 10 validation errors for PostOut FAILED tests/test_post.py::test_get_one_post - pydantic.error_wrappers.ValidationError: 2 validation errors for PostOut If anyone could at least give an opinion of what should be the problem,I would appreciate a lot,this is the second day that I’m trying to find the mistake but I can’t,I even looked at the original code,and it looks that they’re equal