sqlalchemy JSONB column cannot pass validation in Pydantic.BaseModel.from_orm()

[lafrinte]

error msg:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from pydantic_sqlalchemy import sqlalchemy_to_pydantic

engine = create_async_engine('postgresql+asyncpg://postgres:[email protected]/postgres')
session = AsyncSession(engine)

stmt = select(Subjob).where(Subjob.id.in_([1, 2]))
result = await session.execute(stmt)

sj_model = sqlalchemy_to_pydantic(Subjob)
data = [sj_model.from_orm(orm) for orm in result.scalars()]
ValidationError                           Traceback (most recent call last)
<ipython-input-47-b50afa27801d> in <module>
----> 1 data = [sj_model.from_orm(orm) for orm in result.scalars()]

<ipython-input-47-b50afa27801d> in <listcomp>(.0)
----> 1 data = [sj_model.from_orm(orm) for orm in result.scalars()]

/usr/local/lib/python3.8/dist-packages/pydantic/main.cpython-38-x86_64-linux-gnu.so in pydantic.main.BaseModel.from_orm()

ValidationError: 1 validation error for Subjob
hosts
  value is not a valid dict (type=type_error.dict)

my sqlalchemy model code of Subjob

...
class Subjob(PK, Job):
    ...
    hosts = Column(JSONB, nullable=False)
    ...

the hosts is a JSONB type column which will contains python data format like below:

[
    {
        "ip": "172.17.0.2", 
        "port": 5432, 
        "user": "postgres", 
        "password": "postgres"
   },
    {
        "ip": "172.17.0.3", 
        "port": 5432, 
        "user": "postgres", 
        "password": "postgres"
   }
]

the sj_model translate the JSONB as dict type in python which may cause this error.

In [49]: sj_model.__fields__['hosts']
Out[49]: ModelField(name='hosts', type=dict, required=True)

enable List and Dict may fix this

[ljluestc]

The issue you're facing is because the JSONB column hosts in your SQLAlchemy model Subjob is being treated as a dictionary by the Pydantic model generated by sqlalchemy_to_pydantic. To fix this, you need to customize the Pydantic model for the hosts field to ensure it can handle the JSONB data correctly.

Here's the complete updated code with the necessary fixes:

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from pydantic_sqlalchemy import sqlalchemy_to_pydantic
from pydantic import BaseModel

# Define your SQLAlchemy model here (Subjob)
# ...

# Create an Async SQLAlchemy engine and session
engine = create_async_engine('postgresql+asyncpg://postgres:[email protected]/postgres')
session = AsyncSession(engine)

# Define a Pydantic model for the JSONB column 'hosts'
class HostModel(BaseModel):
    ip: str
    port: int
    user: str
    password: str

# Generate the Pydantic model for Subjob
SubjobPydantic = sqlalchemy_to_pydantic(Subjob)

# Fetch data from the database
stmt = select(Subjob).where(Subjob.id.in_([1, 2]))
result = await session.execute(stmt)
data = [SubjobPydantic.from_orm(orm) for orm in result.scalars()]

# Cleanup session
await session.close()

# Print the fetched data
print(data)

In this code, we define a separate Pydantic model HostModel to handle the hosts JSONB data. Then, when generating the Pydantic model for Subjob, we configure the hosts field to use the HostModel by specifying schema_extra:

class Subjob(PK, Job):
    ...
    hosts = Column(JSONB, nullable=False, schema=schema_extra)
    ...

With this approach, the JSONB data will be correctly validated and parsed into the desired structure when using Pydantic models.