Additional support for JSONB Fields

[shepilov-vladislav]

Can https://github.com/collerek/ormar/issues/434 be fixed with your project?

[etimberg]

@shepilov-vladislav I converted your test case to use Postgres and I get the same error

    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,
    ) -> 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:
>               raise ValidationError(errors, field.type_)
E               pydantic.error_wrappers.ValidationError: 1 validation error for Thing_EKE
E               response -> json_field
E                 JSON object must be str, bytes or bytearray (type=type_error.json)

ormar-postgres-extensions.venv/lib/python3.9/site-packages/fastapi/routing.py:137: ValidationError

[etimberg]

My test code looked like

import uuid
from typing import List

import pydantic
import pytest
from fastapi import FastAPI
from httpx import AsyncClient

import ormar
from ormar_postgres_extensions.fields import JSONB
from tests.database import database, metadata

app = FastAPI()


class Thing(ormar.Model):
    class Meta:
        metadata = metadata
        database = database
        tablename = "things"

    id: uuid.UUID = ormar.UUID(primary_key=True, default=uuid.uuid4)
    name: str = ormar.Text(default="")
    json_field: pydantic.Json = JSONB()


@app.get("/things", response_model=List[Thing])
async def read_things():
    print("in request")
    return await Thing.objects.order_by("name").all()


ResponseThing = Thing.get_pydantic(exclude={"id": ...})


@app.get("/things/{id}", response_model=ResponseThing)
async def read_things(id: uuid.UUID):
    return await Thing.objects.get(pk=id)


@pytest.fixture()
async def test_client_async():
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client


@pytest.mark.asyncio
async def test_read_main(db, test_client_async):
    thing1 = await Thing.objects.create(name="thing1", json_field={"1": "1", "2": 2})

    response = await test_client_async.get("/things")
    assert response.status_code == 200
    assert response.json() == [{'id': str(thing1.id), 'name': 'thing1', 'json_field': {'1': '1', '2': 2}}]

    response = await test_client_async.get(f"/things/{thing1.id}")  # Failed here
    #     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,
    #     ) -> 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:
    # >               raise ValidationError(errors, field.type_)
    # E               pydantic.error_wrappers.ValidationError: 1 validation error for Thing_OSI
    # E               response -> json_field
    # E                 JSON object must be str, bytes or bytearray (type=type_error.json)
    assert response.status_code == 200
    assert response.json() == {'name': 'thing1', 'json_field': {'1': '1', '2': 2}}

[Dzhalal1]

what about Thing.objects.filter(json_field__icontains='1')?

[etimberg]

@Dzhalal1 that's not currently supported, but it should be possible to build that functionality in the same way that #9 works.

[Dzhalal1]

Thank you @etimberg

[etimberg]

@Dzhalal1 I implemented the other JSONB operators (contains, contained_by, has_key, has_all, has_any) in v2.2.0