Pydantic: Access Relationship of SQLAlchemy in pydantic Models

[ankithherle]

I am using SQLAlchemy orm and Pydantic with Fast API framework.

My SQLAlchemy models looks like

class Employee(base):

    __tablename__ = "employees"


class Address(base):

    __tablename__ = "addresses"

    employee = sqlalchemy.orm.relationship(
        "Employee",
        backref=sqlalchemy.orm.backref("addresses", lazy="dynamic"),
        foreign_keys="addresses.employee_id",
    )

And My Pydantic Schema looks like as below

class Employee(pydantic.BaseModel):
    id: int
    name: str
    type: str


class Address(pydantic.BaseModel):
    id: int
    city: str
    country: str
    email: str
    first_name: str
    last_name: str
    phone_number: str
    state: str
    employee: Employee

    class Config:
        orm_mode = True

But when I try to access address.employee, it says object of type 'AppenderQuery' has no len() (type=type_error)

So my question is How to access sqlalchemy relationships in pydantic ?

[samuelcolvin]

What is AppenderQuery?

[ankithherle]

AppendQuery is a Sqlachemy relationship-related object.

I want to know how to get the data from sqlalchemy.orm.relationship ?

[samuelcolvin]

I don't use sqlalachemy much. I think ORMs are a conceptual mistake.

Have you looked at orm_mode? https://pydantic-docs.helpmanual.io/usage/models/#orm-mode-aka-arbitrary-class-instances

Alternatively your best option is probably to add a validator to that field that converts the AppenderQuery into a dict that pydantic knows how to understand.

[rusnyder]

@ankithherle While orm_mode isn't aware of Query types (i.e. - lazy relationship properties from SQLAlchemy ORM), I'd suggest combining two things:

1. Set orm_mode = True in your model's config
2. Add a validator that evaluates lazy relationships you're attempting to serialize

I've done so in an OrmBase class in the top of my pydantic model hierarchy for all models that I use to interact w/ SQLAlchemy. You can absolutely apply this pattern in a more targeted fashion, but I'll share my setup which has worked well for me.

Here's a working example:

from pydantic import BaseModel, validator
from sqlalchemy.orm import Query


class OrmBase(BaseModel):
    # Common properties across orm models
    id: int

    # Pre-processing validator that evaluates lazy relationships before any other validation
    # NOTE: If high throughput/performance is a concern, you can/should probably apply
    #       this validator in a more targeted fashion instead of a wildcard in a base class.
    #       This approach is by no means slow, but adds a minor amount of overhead for every field
    @validator("*", pre=True)
    def evaluate_lazy_columns(cls, v):
        if isinstance(v, Query):
            return v.all()
        return v

    class Config:
        orm_mode = True


class Employee(OrmBase):
    name: str
    type: str


class Address(OrmBase):
    city: str
    country: str
    email: str
    first_name: str
    last_name: str
    phone_number: str
    state: str
    employee: Employee

[wangxiaolong100]

I use relationships.lazy="joined" to eager loading relationship.

[eytanschulman]

@rusnyder In your provided example, the two classes inherit BaseModel, not a base of SQLAlchemy. Therefore there would not be any field of type Query.

In this case, how do you integrate the SQLAlchemy models with the Pydantic ones?

@ankithherle While orm_mode isn't aware of Query types (i.e. - lazy relationship properties from SQLAlchemy ORM), I'd suggest combining two things:

1. Set orm_mode = True in your model's config
2. Add a validator that evaluates lazy relationships you're attempting to serialize

I've done so in an OrmBase class in the top of my pydantic model hierarchy for all models that I use to interact w/ SQLAlchemy. You can absolutely apply this pattern in a more targeted fashion, but I'll share my setup which has worked well for me.

Here's a working example:

from pydantic import BaseModel, validator
from sqlalchemy.orm import Query


class OrmBase(BaseModel):
    # Common properties across orm models
    id: int

    # Pre-processing validator that evaluates lazy relationships before any other validation
    # NOTE: If high throughput/performance is a concern, you can/should probably apply
    #       this validator in a more targeted fashion instead of a wildcard in a base class.
    #       This approach is by no means slow, but adds a minor amount of overhead for every field
    @validator("*", pre=True)
    def evaluate_lazy_columns(cls, v):
        if isinstance(v, Query):
            return v.all()
        return v

    class Config:
        orm_mode = True


class Employee(OrmBase):
    name: str
    type: str


class Address(OrmBase):
    city: str
    country: str
    email: str
    first_name: str
    last_name: str
    phone_number: str
    state: str
    employee: Employee

[rusnyder]

@rusnyder In your provided example, the two classes inherit BaseModel, not a base of SQLAlchemy. Therefore there would not be any field of type Query.

In this case, how do you integrate the SQLAlchemy models with the Pydantic ones?

We're not directly integrating the Pydantic models into the same class as the SQLAlchemy models, not would I recommend this as SQLAlchemy expresses its properties in-terms-of DB constraints, whereas Pydantic expresses its properties in terms of serialization/deserialization.

In my posted example, you'd still create separate Pydantic models and SQLALchemy models, then use Pydantic's BaseModel.from_orm(sqlalchemy_obj) to convert the SQLAlchemy object into a Pydantic one. The custom validator in my shared example just "enhances" the from_orm method to also be able to handle lazy relationships, remembering that @validator methods evaluate fields as they are being deserialized into the Pydantic model (i.e. - during from_orm).

Here's a more modified example of the SQLAlchemy-to-Pydantic bits (Updated base example to include a proper one-to-many relationship where, without the custom @validator, the Employee.from_orm() call would choke on the lazy aliases field):

# ########################################### #
#     Step 1: Define SQLAlchemy Models        #
# ########################################### #
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class EmployeeTable(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    aliases = relationship("EmailAliasTable", lazy="dynamic")  # lazy-loading relationship

class EmailAliasTable(Base):
    __tablename__ = 'email_aliases'
    id = Column(Integer, primary_key=True)
    email = Column(String)
    employee_id = Column(Integer, ForeignKey(EmployeeTable.id), nullable=True)
    employee = relationship("EmployeeTable")


# ########################################### #
#     Step 2: Define Pydantic Schemas         #
# ########################################### #
import typing
from pydantic import BaseModel, validator
from sqlalchemy.orm import Query

class OrmBase(BaseModel):
    # Common properties across orm models
    id: int

    # Pre-processing validator that evaluates lazy relationships before any other validation
    # NOTE: If high throughput/performance is a concern, you can/should probably apply
    #       this validator in a more targeted fashion instead of a wildcard in a base class.
    #       This approach is by no means slow, but adds a minor amount of overhead for every field
    @validator("*", pre=True)
    def evaluate_lazy_columns(cls, v):
        if isinstance(v, Query):
            return v.all()
        return v

    class Config:
        orm_mode = True

class EmailAlias(OrmBase):
    email: str

class Employee(OrmBase):
    name: str
    aliases: typing.List[EmailAlias]


# ########################################### #
#     Step 3: Putting it all together         #
# ########################################### #
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Create DB Session
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# Create an employee and an alias in the DB
employee = EmployeeTable(name='Bob Barker')
alias1 = EmailAliasTable(email='[email protected]', employee=employee)
alias2 = EmailAliasTable(email='[email protected]', employee=employee)
session.add_all([employee, alias1, alias2])
session.commit()

# Deserialize the Employee into a Pydantic object
pydantic_employee = Employee.from_orm(employee)
print(pydantic_employee.json())
# {"id": 1, "name": "Bob Barker", "aliases": [{"id": 1, "email": "[email protected]"}, {"id": 2, "email": "[email protected]"}]}