Pydantic: Pydantic and SQLAlchemy

[meandus]

Bug

I have a model Pydantic :

class Brand(BaseModel):
    id: int = None
    name: Optional[str] = None

and a model for SQLAlchemy

class Brand(Base):
    __tablename__ = "brand"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)
    devices = relationship("Device", back_populates="brand")
    created_at = Column(DateTime, default=datetime.datetime.now)
    updated_at = Column(DateTime, default=datetime.datetime.now)

    def __repr__(self):
        return "<Brand(id='%d', name='%s')>" % (self.id, self.name)

and it works like a charm

Pydantic for Nested

Class Device(BaseModel)
id: int = None
brand: Brand = None

when i get brand, i receive the object but i have also _sa_instance_state and pydantic send a Validation Error , so i move from Class to Pydantic Class (Test) , SQLAlchemy model does not work properly because AttributeError for _sa_instance_state.

Do you have any idea to avoid issue for Pydantic Validation for this specific key/value ?

Thanks in advance,
Rémy

[samuelcolvin]

I'm afraid I don't understand your question.

Please explain more clearly and try to format your code snippets so they can be read easily.

[meandus]

Exception has occurred: ValidationError
1 validation error
brand
  value is not a valid dict (type=type_error.dict)

[meandus]

it's a bit complex to make a full example but anyway i xwill do

[meandus]

app/db/_ _ init _ _.py

from sqlalchemy.ext.declarative import declarative_base, declared_attr
import os
import sys
import importlib
import glob
from app.api.utils.logging import logger
from pathlib import Path


class CustomBase(object):
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)

main.py

from sqlalchemy.orm import relationship, scoped_session, sessionmaker
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from app.db import Base
from pydantic import BaseModel
from pprint import pprint
import random
import string


class Test(Base):
    __tablename__ = "test"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)

    def __repr__(self):
        return "<Test(id='%d', name='%s')>" % (self.id, self.name)


class Data(Base):
    __tablename__ = "data"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, unique=True, index=True)
    test_id = Column(Integer, ForeignKey("test.id"))
    test = relationship("Test")

    def __repr__(self):
        return "<Data(id='%d', name='%s')>" % (self.id, self.name)


class PTest(BaseModel):
    id: int = None
    name: str = None


class PData(BaseModel):
    id: int = None
    name: str = None
    test: PTest = None

if __name__ == "__main__":

    def random_lower_string(k: int = 32):
        return "".join(random.choices(string.ascii_letters, k=k))

    def create_test(db_session, *, test_in: PTest) -> PTest:
        test = Test(name=test_in.name)
        db_session.add(test)
        db_session.commit()
        db_session.refresh(test)
        return test

    def get_test(db_session, *, test_id: int) -> PTest:
        return db_session.query(Test).filter(Test.id == test_id).first()

    def create_data(db_session, *, data_in: PData) -> PData:
        data = Data(name=data_in.name, test=data_in.test)
        db_session.add(data)
        db_session.commit()
        db_session.refresh(data)
        return data

    def get_data(db_session, *, data_id: int) -> PTest:
        return db_session.query(Data).filter(Data.id == data_id).first()

    engine = create_engine(f"postgresql://docker:docker@db/connectivity")
    db_session = scoped_session(
        sessionmaker(autocommit=False, autoflush=False, bind=engine)
    )

    test_in = PTest(name=random_lower_string(10))
    test = create_test(db_session, test_in=test_in)
    test = get_test(db_session, test_id=test.id)
    pprint(test)

    data_in = PData(name=random_lower_string(10), test=test)
    data = create_data(db_session, data_in=data_in)
    data = get_data(db_session, data_in=data_in)
    pprint(data)

output:

<Test(id='14', name='RjhBWgeQRm')>
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7efd5bb804a8>,
 'id': 14,
 'name': 'RjhBWgeQRm'}
Traceback (most recent call last):
  File "main2.py", line 77, in <module>
    data_in = PData(name=random_lower_string(10), test=test)
  File "/usr/local/lib/python3.7/site-packages/pydantic/main.py", line 232, in __init__
    object.__setattr__(self, '__values__', self._process_values(data))
  File "/usr/local/lib/python3.7/site-packages/pydantic/main.py", line 432, in _process_values
    return validate_model(self, input_data)  # type: ignore
  File "/usr/local/lib/python3.7/site-packages/pydantic/main.py", line 627, in validate_model
    raise ValidationError(errors)
pydantic.error_wrappers.ValidationError: 1 validation error
test
  value is not a valid dict (type=type_error.dict)

[meandus]

First pprint is Test and second is pprint(Test.__dict__)

Third pprint is failed with data validation error for object class Data

[samuelcolvin]

data_in = PData(name=random_lower_string(10), test=test)

The object you're passing to test isn't a valid dictionary or a pydantic model, or a class that implements __get_validators__.

You can either set arbitrary_types_allowed in config or define __get_validators__ as per #501

[meandus]

something like extra in model.Config could help ?

[meandus]

class PTest(BaseModel):
    id: int = None
    name: str = None

    class Config:
        arbitrary_types_allowed = True
        extra = Extra.allow


class PData(BaseModel):
    id: int = None
    name: str = None
    test: PTest = None

    class Config:
        arbitrary_types_allowed = True
        extra = Extra.allow

still the same:

<Test(id='28', name='LIkqlGcIOe')>
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7efd393aa4e0>,
 'id': 28,
 'name': 'LIkqlGcIOe'}
Traceback (most recent call last):
  File "main2.py", line 89, in <module>
    data_in = PData(name=random_lower_string(10), test=test)
  File "/usr/local/lib/python3.7/site-packages/pydantic/main.py", line 232, in __init__
    object.__setattr__(self, '__values__', self._process_values(data))
  File "/usr/local/lib/python3.7/site-packages/pydantic/main.py", line 432, in _process_values
    return validate_model(self, input_data)  # type: ignore
  File "/usr/local/lib/python3.7/site-packages/pydantic/main.py", line 627, in validate_model
    raise ValidationError(errors)
pydantic.error_wrappers.ValidationError: 1 validation error
test
  value is not a valid dict (type=type_error.dict)

currently installed: pydantic==0.23

[samuelcolvin]

The thing you've passed to test isn't an instance of PTest or a dictionary.

[tbert12]

If you are returning an ORM instance you should set orm_mode = True

class AnItem(BaseModel):
    title: str
    description: str = None
    class Config:
        orm_mode = True

Without orm_mode, if you returned a SQLAlchemy model from your path operation, it wouldn't include the relationship data.

From: sql databases