Skip to content

Instantly share code, notes, and snippets.

@zzzeek
Last active June 7, 2023 20:42
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save zzzeek/3c027db82a4222d01f71fe503228dfc5 to your computer and use it in GitHub Desktop.
Save zzzeek/3c027db82a4222d01f71fe503228dfc5 to your computer and use it in GitHub Desktop.
issues with pydantic dataclasses / sqlalchemy
from __future__ import annotations
from typing import TYPE_CHECKING
import pydantic.dataclasses
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
class Base(
MappedAsDataclass,
DeclarativeBase,
dataclass_callable=pydantic.dataclasses.dataclass,
):
if not TYPE_CHECKING:
# workaround for problem 1 below
id = Column(Integer, primary_key=True)
class B(Base):
__tablename__ = "b"
# problem 1 - there is no way to use a pydantic dataclass field
# with init=False that keeps the value totally unset. this field
# fails validation. if we add default=None to pass validation, SQLAlchemy
# does not use the server side primary key generator.
# SQLAlchemy will likely need to add a new default value DONT_SET to
# work around this
# id: Mapped[int | None] = mapped_column(primary_key=True, init=False)
a_id: Mapped[int | None] = mapped_column(
ForeignKey("a.id"), init=False, default=None
)
class A(Base):
__tablename__ = "a"
# also problem 1 here
# id: Mapped[int | None] = mapped_column(primary_key=True, init=False)
data: Mapped[str]
bs: Mapped[list[B]] = relationship("B")
e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
with Session(e) as s:
# problem 2. Pydantic lets the class init, then validators
# come in and *rewrite* the collections, then assign them on __dict__.
# patch that fixes this specific case here, however there can be many
# more since pydantic writes into `__dict__` quite a lot
a1 = A(data="a1", bs=[B(), B()])
s.add(a1)
s.commit()
@zzzeek
Copy link
Author

zzzeek commented Jun 7, 2023

patch for issue 2, which below is stated against the v1.10.8 tag:

diff --git a/pydantic/dataclasses.py b/pydantic/dataclasses.py
index 86bad1e6..c9e7163d 100644
--- a/pydantic/dataclasses.py
+++ b/pydantic/dataclasses.py
@@ -423,7 +423,10 @@ def _dataclass_validate_values(self: 'Dataclass') -> None:
     d, _, validation_error = validate_model(self.__pydantic_model__, input_data, cls=self.__class__)
     if validation_error:
         raise validation_error
-    self.__dict__.update(d)
+
+    for k, v in d.items():
+        setattr(self, k, v)
+    #self.__dict__.update(d)
     object.__setattr__(self, '__pydantic_initialised__', True)
 

@tiangolo
Copy link

tiangolo commented Jun 7, 2023

I'm not sure what is not working, does that raise an exception? Maybe a couple of asserts could help to verify your expected behavior. I just tried with Pydantic v2, updating it to what I imagine you originally wanted (without the workarounds):

from __future__ import annotations

import pydantic.dataclasses

from sqlalchemy import Column, select
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session


class Base(
    MappedAsDataclass,
    DeclarativeBase,
    dataclass_callable=pydantic.dataclasses.dataclass,
):
    pass
    # if not TYPE_CHECKING:
    #     # workaround for problem 1 below
    #     id = Column(Integer, primary_key=True)


class B(Base):
    __tablename__ = "b"

    # problem 1 - there is no way to use a pydantic dataclass field
    # with init=False that keeps the value totally unset.  this field
    # fails validation.  if we add default=None to pass validation, SQLAlchemy
    # does not use the server side primary key generator.
    # SQLAlchemy will likely need to add a new default value DONT_SET to
    # work around this
    id = Column(Integer, primary_key=True)

    a_id: Mapped[int | None] = mapped_column(
        ForeignKey("a.id"), init=False, default=None
    )


class A(Base):
    __tablename__ = "a"

    # also problem 1 here
    id = Column(Integer, primary_key=True)

    data: Mapped[str]
    bs: Mapped[list[B]] = relationship("B")


e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
with Session(e) as s:
    # problem 2.  Pydantic lets the class init, then validators
    # come in and *rewrite* the collections, then assign them on __dict__.
    # patch that fixes this specific case here, however there can be many
    # more since pydantic writes into `__dict__` quite a lot
    a1 = A(data="a1", bs=[B(), B()])
    s.add(a1)
    s.commit()


s2 = Session(e)
ap = s2.execute(select(A)).scalars().first()
print(ap)
# A(data='a1', bs=[B(a_id=1), B(a_id=1)])
print(ap.id)
# 1
for b in ap.bs:
    print(b.id)
# 1
# 2

This seems to work, right?

@zzzeek
Copy link
Author

zzzeek commented Jun 7, 2023

havent tried with pydantic 2.0. they would have to be doing things pretty differently. i am working up a fix for issue number one on our end.

@CaselIT
Copy link

CaselIT commented Jun 7, 2023

Just tried and the following works on pydantic 2

from __future__ import annotations

import pydantic
import pydantic.dataclasses

from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import MappedAsDataclass
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session

assert pydantic.__version__ >= "2"

class Base(
    MappedAsDataclass,
    DeclarativeBase,
    dataclass_callable=pydantic.dataclasses.dataclass,
):
    pass

class B(Base):
    __tablename__ = "b"

    id: Mapped[int] = mapped_column(primary_key=True, init=False)

    a_id: Mapped[int | None] = mapped_column(
        ForeignKey("a.id"), init=False, default=None
    )

class A(Base):
    __tablename__ = "a"

    id: Mapped[int] = mapped_column(primary_key=True, init=False)

    data: Mapped[str]
    bs: Mapped[list[B]] = relationship("B")

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)
with Session(e) as s:
    a1 = A(data="a1", bs=[B(), B()])
    s.add(a1)
    s.commit()

with Session(e) as s:
    a = s.scalars(select(A)).one()
    assert a.data == "a1"
    assert len(a.bs) == 2

i am working up a fix for issue number one on our end.

I guess we can just wait instead

@zzzeek
Copy link
Author

zzzeek commented Jun 7, 2023

how does the int id with init=False work? does it only validate arguments that were actually passed ? (because hooray if so?)

@CaselIT
Copy link

CaselIT commented Jun 7, 2023

how does the int id with init=False work? does it only validate arguments that were actually passed ? (because hooray if so?)

no clue. I guess it does?

@CaselIT
Copy link

CaselIT commented Jun 7, 2023

I think the best option at the moment is to wait for v2 to be released and at that point look if anything is still needed. At the moment is seems no, but it's a beta the current v2 version

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment