Uh oh!
There was an error while loading. Please reload this page.
✨ Add PydanticJSONB TypeDecorator for Automatic Pydantic Model Serialization in SQLModel - #1324
✨ Add PydanticJSONB TypeDecorator for Automatic Pydantic Model Serialization in SQLModel#1324amanmibra wants to merge 27 commits into
PydanticJSONB TypeDecorator for Automatic Pydantic Model Serialization in SQLModel#1324Conversation
I like this approach. There seem to me at the moment 2 things that would still need to be added.
What about the following? fromtypingimportAny, Type, TypeVar, get_argsfrompydanticimportBaseModelfromsqlalchemyimporttypesfromsqlalchemy.dialects.postgresqlimportJSONB# for Postgres JSONBBaseModelType=TypeVar("BaseModelType", bound=BaseModel)
classPydanticJSONB(types.TypeDecorator): # type: ignore"""Custom type to automatically handle Pydantic model serialization."""impl=JSONB# use JSONB type in Postgres (fallback to JSON for others)cache_ok=True# allow SQLAlchemy to cache resultsdef__init__(self, model_class: Type[BaseModel] |Type[list[BaseModelType]], *args, **kwargs):
super().__init__(*args, **kwargs)
self.model_class=model_class# Pydantic model class to usedefprocess_bind_param(self, value: Any, dialect) ->list[dict] |dict|None: # noqa: ANN401, ARG002, ANN001ifisinstance(value, BaseModel):
returnvalue.model_dump(mode="json")
ifisinstance(value, list):
return [m.model_dump(mode="json") forminvalue]
returnvaluedefprocess_result_value(self, value: Any, dialect) ->list[BaseModel] |BaseModel|None: # noqa: ANN401, ARG002, ANN001# Called when loading from DB: convert dict to Pydantic model instanceifisinstance(value, dict):
returnself.model_class.model_validate(value) # type: ignoreifisinstance(value, list):
return [get_args(self.model_class)[0].model_validate(v) forvinvalue]
returnNone |
+1 from @DaanRademaker 's comment. This PydanticJSONB implementation needs to support |
amanmibra
commented
Mar 17, 2025
@Seluj78 I hesitate because when process_result_value returns a dict, additional logic is needed to distinguish between a basic dictionary and a structured collection of Pydantic models. This could introduce ambiguity or unexpected behavior in model validation and serialization. |
amanmibra
commented
Mar 17, 2025
Actually, maybe the key signifier can come from the |
Seluj78
commented
Mar 17, 2025
I see what you mean. I was trying to avoid an extra step by just having The main problem you'd need to tacle anyway, no matter if you choose to support this or not is the mutability and assignment detection. It was a nightmare to try and get working and I failed on my end |
amanmibra
commented
Mar 17, 2025
No, it actually makes sense, sorry, I had to think about for a minute. Take a look at my most recent commit. I manage it by, just like with If you want, throw this at the bottom and run if__name__=="__main__":
fromtypingimportDict, ListfrompydanticimportBaseModelfromsqlalchemyimportColumnfromsqlmodelimportField, SQLModel# Import Fieldfromsqlmodel.sql.sqltypesimportPydanticJSONB# Define some Pydantic modelsclassAddress(BaseModel):
street: strcity: strclassUser(BaseModel):
name: strage: int# 1. Single Model ExampleclassPersonTable(SQLModel, table=True):
id: int|None=Field(default=None, primary_key=True)
# Use Field instead of Column, and wrap Address with PydanticJSONBaddress: Address=Field(sa_column=Column(PydanticJSONB(Address)))
# 2. List of Models ExampleclassTeamTable(SQLModel, table=True):
id: int|None=Field(default=None, primary_key=True)
members: List[User] =Field(sa_column=Column(PydanticJSONB(List[User])))
# 3. Dictionary of Models ExampleclassCompanyTable(SQLModel, table=True):
id: int|None=Field(default=None, primary_key=True)
employees: Dict[str, User] =Field(
sa_column=Column(PydanticJSONB(Dict[str, User]))
)
# Test instancesperson=PersonTable(address=Address(street="123 Main St", city="Boston"))
print("Person:")
print(person)
team=TeamTable(members=[User(name="Alice", age=30), User(name="Bob", age=25)])
print("\nTeam:")
print(team)
company=CompanyTable(
employees={"alice": User(name="Alice", age=30), "bob": User(name="Bob", age=25)}
)
print("\nCompany:")
print(company)
|
DaanRademaker
commented
Mar 18, 2025
Nice improvements! I think there is 1 more interesting usecase. Let's say you would use the insert statement from the postgres dialect, the process bind param will be called with a dictionary that might not be jsonable. It would probably be useful to convert any dictionary to jsonable using pydantic to_jsonable function. fromdatetimeimportdatefromdecimalimportDecimalfromtypingimportAny, Dict, List, Type, TypeVar, get_argsfromuuidimportUUIDfrompydanticimportBaseModelfrompydantic_coreimportto_jsonable_pythonfromsqlalchemyimporttypesfromsqlalchemy.dialects.postgresqlimport (
JSONB, # for Postgres JSONBinsert,
)
fromsqlmodelimportField, SQLModelBaseModelType=TypeVar("BaseModelType", bound=BaseModel)
# Add to_jsonable_pythonclassPydanticJSONB(types.TypeDecorator): # type: ignore"""Custom type to automatically handle Pydantic model serialization."""impl=JSONB# use JSONB type in Postgres (fallback to JSON for others)cache_ok=True# allow SQLAlchemy to cache resultsdef__init__(
self,
model_class: Type[BaseModelType] |Type[list[BaseModelType]] |Type[Dict[str, BaseModelType]],
*args,
**kwargs,
):
super().__init__(*args, **kwargs)
self.model_class=model_class# Pydantic model class to usedefprocess_bind_param(self, value: Any, dialect) ->dict|list[dict] |None: # noqa: ANN401, ARG002, ANN001ifvalueisNone:
returnNoneifisinstance(value, BaseModel):
returnvalue.model_dump(mode="json")
ifisinstance(value, list):
return [m.model_dump(mode="json") ifisinstance(m, BaseModel) elseto_jsonable_python(m) forminvalue]
ifisinstance(value, dict):
return {
k: v.model_dump(mode="json") ifisinstance(v, BaseModel) elseto_jsonable_python(v)
fork, vinvalue.items()
}
returnto_jsonable_python(value)
defprocess_result_value(
self, value: Any, dialect
) ->BaseModelType|List[BaseModelType] |Dict[str, BaseModelType] |None:
ifvalueisNone:
returnNoneifisinstance(value, dict):
# If model_class is a Dict type hint, handle key-value pairsifhasattr(self.model_class, "__origin__") andself.model_class.__origin__isdict:
model_class=get_args(self.model_class)[1] # Get the value type (the model)return {k: model_class.model_validate(v) fork, vinvalue.items()}
# Regular case: the whole dict represents a single modelreturnself.model_class.model_validate(value) # type: ignoreifisinstance(value, list):
# If model_class is a List type hintifhasattr(self.model_class, "__origin__") andself.model_class.__origin__islist:
model_class=get_args(self.model_class)[0]
return [model_class.model_validate(v) forvinvalue]
# Fallback case (though this shouldn't happen given our __init__ types)return [self.model_class.model_validate(v) forvinvalue] # type: ignorereturnvalueclassSomeNestedModel(BaseModel):
some_decimal: DecimalclassSomeModel(SQLModel, table=True):
some_uuid: UUID=Field(primary_key=True)
some_date: date=Field(primary_key=True)
some_nested: SomeNestedModel=Field(sa_column=Field(PydanticJSONB(SomeNestedModel)))
item_1=SomeModel(
some_uuid=UUID("123e4567-e89b-12d3-a456-426614174000"),
some_date=date(2022, 1, 1),
some_nested=SomeNestedModel(some_decimal=Decimal("1.23")), #
)
# model_dump(mode="json") cannot be called because else fields like some_date and some_uuid are not correct type# therefore added to_jsonable_python in process_bind_paramnot_yet_jsonable_dict=item_1.model_dump()
stmt=insert(SomeModel).values(not_yet_jsonable_dict)
# statement cannot executed with jsonable dict because it loses datetime, uuid types etcjsonable_dict=item_1.model_dump(mode="json")
stmt_2=insert(SomeModel).values(jsonable_dict) |
amanmibra
commented
Mar 19, 2025
@DaanRademaker Also, I tried your example (make sure your update the fromdatetimeimportdatefromdecimalimportDecimalfrompydanticimportBaseModelfromuuidimportUUIDfromsqlmodelimportColumn, Field, SQLModelclassSomeNestedModel(BaseModel):
some_decimal: DecimalclassSomeModel(SQLModel, table=True):
some_uuid: UUID=Field(primary_key=True)
some_date: date=Field(primary_key=True)
some_nested: SomeNestedModel=Field(
sa_column=Column(PydanticJSONB(SomeNestedModel))
)
item_1=SomeModel(
some_uuid=UUID("123e4567-e89b-12d3-a456-426614174000"),
some_date=date(2022, 1, 1),
some_nested=SomeNestedModel(some_decimal=Decimal("1.23")), #
)
print(item_1) # some_uuid=UUID('123e4567-e89b-12d3-a456-426614174000') some_date=datetime.date(2022, 1, 1) some_nested=SomeNestedModel(some_decimal=Decimal('1.23')) |
PydanticJSONB TypeDecorator for Automatic Pydantic Model Serialization in SQLModelPydanticJSONB TypeDecorator for Automatic Pydantic Model Serialization in SQLModelsvlandeg
commented
Mar 19, 2025
Hi all, Thanks for the contribution and it's great to see this level of engagement! Just as a maintenance note, I'll put this in draft while the CI is failing. Feel free to mark as "Ready for review" when it's green! |
amanmibra
commented
Mar 21, 2025
🫡 |
I took some time to create a fully reproducible example to show the error I am getting. The issue seems to be with difference between using a sync adapter (psycopg2) or async adapter (asyncpg). Asyncpg cannot deal with already jsonable python dict before hitting the process_bind_param function. It seems asyncpg does stricter type handling during parameter substitution and does not convert to the correct type as psycopg2 does. By adding the to_jsonable_python in the process_bind_param function we are able to pass an not yet jsonable dict. Which resolves this issue with asyncpg. asyncdeftest_reproducable_example():
fromdatetimeimportdate, datetimefromdecimalimportDecimalfromtypingimportAny, Dict, List, Type, TypeVar, get_argsfromuuidimportUUIDfrompydanticimportBaseModelfromsqlalchemyimporttypesfromsqlalchemy.dialects.postgresqlimport (
JSONB, # for Postgres JSONBinsert,
)
fromsqlmodelimportColumn, Field, Session, SQLModel, create_engine, selectfromsqlmodel.ext.asyncio.sessionimportAsyncSessionfromsqlalchemy.ext.asyncioimportcreate_async_engineBaseModelType=TypeVar("BaseModelType", bound=BaseModel)
# fill in your database connection details herepostgres_url="postgresql+psycopg2://user:password@localhost:5432/dbname"postgres_url_async="postgresql+asyncpg://user:password@localhost:5432/dbname"async_engine=create_async_engine(postgres_url_async)
sync_engine=create_engine(postgres_url)
classPydanticJSONB(types.TypeDecorator): # type: ignore"""Custom type to automatically handle Pydantic model serialization."""impl=JSONB# use JSONB type in Postgres (fallback to JSON for others)cache_ok=True# allow SQLAlchemy to cache resultsdef__init__(
self,
model_class: Type[BaseModelType] |Type[list[BaseModelType]] |Type[Dict[str, BaseModelType]],
*args,
**kwargs,
):
super().__init__(*args, **kwargs)
self.model_class=model_class# Pydantic model class to usedefprocess_bind_param(self, value: Any, dialect) ->dict|list[dict] |None: # noqa: ANN401, ARG002ifvalueisNone:
returnNoneifisinstance(value, BaseModel):
returnvalue.model_dump(mode="json")
ifisinstance(value, list):
return [m.model_dump(mode="json") ifisinstance(m, BaseModel) elsemforminvalue]
ifisinstance(value, dict):
return {k: v.model_dump(mode="json") ifisinstance(v, BaseModel) elsevfork, vinvalue.items()}
returnvaluedefprocess_result_value(
self, value: Any, dialect
) ->BaseModelType|List[BaseModelType] |Dict[str, BaseModelType] |None:
ifvalueisNone:
returnNoneifisinstance(value, dict):
# If model_class is a Dict type hint, handle key-value pairsifhasattr(self.model_class, "__origin__") andself.model_class.__origin__isdict:
model_class=get_args(self.model_class)[1] # Get the value type (the model)return {k: model_class.model_validate(v) fork, vinvalue.items()}
# Regular case: the whole dict represents a single modelreturnself.model_class.model_validate(value) # type: ignoreifisinstance(value, list):
# If model_class is a List type hintifhasattr(self.model_class, "__origin__") andself.model_class.__origin__islist:
model_class=get_args(self.model_class)[0]
return [model_class.model_validate(v) forvinvalue]
# Fallback case (though this shouldn't happen given our __init__ types)return [self.model_class.model_validate(v) forvinvalue] # type: ignorereturnvalueclassSomeNestedModel(BaseModel):
some_decimal: DecimalclassSomeModel(SQLModel, table=True):
some_uuid: UUID=Field(primary_key=True)
some_date: date=Field(primary_key=True)
some_nested: SomeNestedModel=Field(sa_column=Column(PydanticJSONB(SomeNestedModel)))
SQLModel.metadata.create_all(sync_engine)
item_1=SomeModel(
some_uuid=UUID("123e4567-e89b-12d3-a456-426614174000"),
some_date=date(2022, 1, 1),
some_nested=SomeNestedModel(some_decimal=Decimal("1.23")),
)
items_mode_json= [item_1.model_dump(mode="json")]
items_non_json_mode= [item_1.model_dump()]
stmt_json_mode=insert(SomeModel).values(items_mode_json)
stmt_non_json_mode=insert(SomeModel).values(items_non_json_mode)
# Psycopg2 engine causing no problems!withSession(sync_engine) assession:
session.execute(stmt_json_mode)
session.commit()
# This fails with statement error, the PydanticJSONB type decorator process_bind_param is called# With a dict that does not have json serializable values (Decimal is not serializable)try:
withSession(sync_engine) assession:
session.execute(stmt_non_json_mode)
session.commit()
exceptExceptionase:
print(e)
try:
# This fails with DBAPIError todordinal of the date type, asyncpg engine cannot deal with# Date values already being converted to stringasyncwithAsyncSession(async_engine) assession:
awaitsession.execute(stmt_json_mode)
awaitsession.commit()
exceptExceptionase:
print(e)
try:
# This fails with StatementError the PydanticJSONB type decorator process_bind_param is called# With a dict that does not have json serializable values (Decimal is not serializable)asyncwithAsyncSession(async_engine) assession:
awaitsession.execute(stmt_non_json_mode)
awaitsession.commit()
exceptExceptionase:
print(e) |
… for non-BaseModel types in lists and dictionaries
amanmibra
commented
Mar 22, 2025
@DaanRademaker Thanks for sending that example! Helped me understand the blocker you were presenting Made the updates to include Only issue now is the linter is yelling at me due to the function returning |
This comment was marked as spam.
This comment was marked as spam.
Any advice in how will be used in a |
This comment was marked as spam.
This comment was marked as spam.
JBorrow
left a comment
There was a problem hiding this comment.
Happy with new docs, needs a core maintainer to review and merge.
This comment was marked as spam.
This comment was marked as spam.
Thanks @amanmibra for this much needed feature. Have you tested alembic migrations with this? Update: The render_item hook in alembic's env.py helped: defrender_item(type_: str, obj: Any, autogen_context: Any) ->Any: # noqa: ANN401, ARG001"""Replace PydanticJSONB with JSONB for migrations."""iftype_=="type"andisinstance(obj, PydanticJSONB):
# Render PydanticJSONB as standard JSONB for migrationsreturn"postgresql.JSONB(astext_type=sa.Text())"returnFalsecontext.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
render_item=render_item,
) |
amanmibra
commented
Aug 6, 2025
I have not! I forgot to address the feedback earlier from @DavidKatz-il
Should I go ahead and just the |
@amanmibra I added support for the following by building on top of your PR. Feel free to include it.
Limitations
|
jonashaag
commented
Oct 22, 2025
To automatically set @classmethoddef__class_getitem__(cls, model_class: type[T]) ->type["PydanticJSONB[T]"]:
"""Create a new subclass with the specified model class."""new_class=type(cls.__name__, (cls,), {"_model_class": model_class, "cache_ok": True})
returnnew_class |
amanmibra
commented
Nov 10, 2025
slmnsh
commented
Nov 17, 2025
alembic migration support stated in #1324 (comment) |
amanmibra
commented
Nov 17, 2025
Ngl, I am personally slammed to add this in myself, is there a way to have the community add that on? Maybe an child PR? |
I don't think maintainers will allow that. EDIT: If you guide how someone can add it in this PR. maybe someone will just do it. |
slmnsh
commented
Feb 20, 2026
@svlandeg is there any update on this? everything is green. |
svlandeg
commented
Feb 22, 2026
Please just stop pinging maintainers here. We have so much on our plate, and managing GH notifications becomes a real struggle, preventing us to work more efficiently on all the open-source tasks we have. This is in our queue but further spamming/pinging certainly won't help boost its priority, to be perfectly honest with you. |
amanmibra
commented
Feb 23, 2026
I appreciate the work you all do, but this was also a fair ask, as it has been 4-5 month since any activity, and it was opened nearly a year ago. @slmnsh is not the author of this PR, just a fellow dev. @slmnsh I just recommend you copy this module directly into your project. That was worked just fine for me over the last year. |
📝 Docs previewLast commit 2f8b894 at: https://1a4915db.sqlmodel.pages.dev Modified Pages |
CHC383
commented
May 11, 2026
@amanmibra Inspired by your work and several others from #63, I have listed two alternative implementations in #63 (comment), please feel free to adopt them. Here are some takeaways:
|
amanmibra
commented
May 13, 2026
@CHC383 if you want, you can branch off of this branch and I can merge it in so it guarantees you get git commit credit for that contribution (in a world were this finally gets merged in lol) |
This pull request has a merge conflict that needs to be resolved. |
CHC383
commented
May 22, 2026
@amanmibra Here you go, please feel free to adjust the changes. |
Description
This PR introduces a
PydanticJSONBto SQLModel, enabling seamless serialization and deserialization of Pydantic models in JSONB columns. This removes the need for manual conversion, allowing SQLModel instances to work directly with Pydantic objects.Why?
Storing Pydantic models in JSONB columns has been a recurring challenge. This PR solves that by automating conversion between Pydantic models and JSON fields.
How?
Benefits
✅ Eliminates manual conversion – No need to wrap dict(**org.config) manually.
✅ Ensures structured storage – Enforces Pydantic validation automatically.
✅ Improves dev experience – Seamless interaction with JSONB fields in SQLModel.
Example Usage of PydanticJSONB in SQLModel
With this PR, you can now store and retrieve Pydantic models in JSONB fields effortlessly.
Define a Pydantic Model
Create & Store Data
Retrieve & Use Data
Result:
✅ No need for OrgConfig(**org.config) – it's already a OrgConfig instance!
✅ Automatic conversion between JSONB and Pydantic models.
This simplifies handling structured configurations in SQLModel, making JSONB storage seamless and ergonomic. 🚀
Related Issues & Discussions
SQLModel Issue #63 – Nested Pydantic models in JSON fields.
Stack Overflow: Writing Pydantic objects into SQLAlchemy JSON columns
GitHub Gist Example – Community implementation of Pydantic models in JSON columns.
This PR makes SQLModel more ergonomic for JSONB storage while maintaining compatibility with SQLAlchemy and Pydantic. 🚀
TODO
Per contribution rules: https://sqlmodel.tiangolo.com/help/#create-a-pull-request
This PR still needs:
Before I do that, I would love to hear your thoughts @tiangolo!