CRUD Sqlalchemy ORM, Pydantic and query on multiple table (like join table)

[tatdatpham]

I have trouble on making DB CRUD with two or multiple table (join table) and this is first time i working with Pydantic and FastAPI.

Sorry if my question is bullshit :'(
I have two database model

Shifdetail.py

class ShiftDetail(Base):
    id = Column(String, primary_key=True, index=True)
    shift_id = Column(String, ForeignKey("shift.id"))
    owner_id = Column(String, ForeignKey("user.id"))
    shift_date = Column(Date, nullable=False)
    create_time = Column(DateTime, nullable=False)
    shifts = relationship("Shift", back_populates="shiftdetails")
    owner = relationship("User", back_populates="shiftdetails")

User.py

class User(Base):
    id = Column(String, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    shiftdetails = relationship("ShiftDetail", back_populates="owner")

EXPECTED data return look like

[
  {
    "shift_id": "5240b998-7919-11ea-8f22-001a7dda7111",
    "owner_id": 56277c6c-7918-11ea-850b-001a7dda7111,
    "shift_date": "2020-04-08",
    "id": "87845423-795f-11ea-a51f-001a7dda7111",
    "create_time": "2020-04-08T13:09:39.067190",
    "owner_email": "admin@email.com"
  }
]

And have ShiftDetail schema look like

class ShiftDetailBase(BaseModel):
    shift_id: Optional[str] = True
    owner_id: Optional[str] = True
    shift_date: Optional[date] = True

class ShiftDetailBaseInDB(ShiftDetailBase):
    id: str

    class Config:
        orm_mode = True

# Additional properties to return via API
class ShiftDetail(ShiftDetailBaseInDB):
    create_time: datetime
    owner_email: str

So how can write CRUD function to return datamodel ShiftDetail (with data get from joining two table ShiftDetail and User)

Because in CRUD template, i saw db_session query from self.model only :'(

def get_multi_by_owner(
        self, db_session: Session, *, owner_id: str, skip=0, limit=100
    ) -> List[ShiftDetail]:
        return (
            db_session.query(self.model)
            .filter(ShiftDetail.owner_id == owner_id)
            .offset(skip)
            .limit(limit)
            .all()
        )

And of course, i got error (ShiftDetail model dont have owner_email field like ShiftDetail schema expected

pydantic.error_wrappers.ValidationError: 1 validation error for ShiftDetail
response -> 0 -> owner_email

[paul121]

You can't simply add a user_email field to that schema and expect it to be populated. Rather,
you need to also create a User schema. Otherwise Pydantic won't now how to connect the two models.

class User(BaseModel):
    class Config:
        orm_mode = True
    
    id: str
    email: str
    hashed_password: str

class ShiftDetailBase(BaseModel)
    # add field for the owner (reference User model)
    # I believe you can remove the owner_id field
    owner: User

The response would then look like:

[
  {
    "shift_id": "5240b998-7919-11ea-8f22-001a7dda7111",
    "owner_id": 56277c6c-7918-11ea-850b-001a7dda7111,
    "shift_date": "2020-04-08",
    "id": "87845423-795f-11ea-a51f-001a7dda7111",
    "create_time": "2020-04-08T13:09:39.067190",
    "owner": {
       "id": "..",
       "email": "admin@email.com",
       "hashed_password": "...",
    }
  }
]

If you want user_email to be returned from that schema you would need to alter the design of your DB or I believe it might be possible with a custom Pydantic root validator - but that's making things more complex and breaks from the ORM model.

[tatdatpham]

@paul121 . Sure, thank for your guide about schema. But i very confuse about making CRUD to return data that i need.

I have some code to return list of ShiftDetail (not include Owner)

def get_multi_by_owner(
        self, db_session: Session, *, owner_id: str, skip=0, limit=100
    ) -> List[ShiftDetail]:
        return (
            db_session.query(self.model)
            .filter(ShiftDetail.owner_id == owner_id)
            .offset(skip)
            .limit(limit)
            .all()
        )

def get_by_id(self, db_session: Session, *, id: str) -> Optional[User]:
        return db_session.query(User).filter(User.id == id).first()

And i dont know to to make data include Owner data ...
Normaly, i get list Shiftdetail, list User ; For LIST and mapping Owner_id with user.id in User; Return new list as expected
Is that best way to do in FastAPI. I dont think so :(

[paul121]

Can we think about the query in the reverse direction? This should return only the ShiftDetails that belong to one user. (not tested) ... eg:

def get_shifts_by_owner_id(
        self, db_session: Session, *, owner_id: str, skip=0, limit=100
    ) -> List[ShiftDetail]:
        return (
            db_session.query(model.User.shiftdetails). # query on the User model, select only the shift details
            .filter(model.User.id == owner_id)
            .offset(skip)
            .limit(limit)
            .all()
        )

[tatdatpham]

Just query on one table only and model User dont have shiftdetails. It doesn't work :'(

[paul121]

hmm, sorry that didn't work @tatdatpham! ☹️ I'm not sure what might be the issue. This likely isn't an issue with FastAPI, though

[tiangolo]

Thanks for the help here @paul121 ! 👏 🍰

@tatdatpham First make sure that your SQLAlchemy query is returning the data that you want. Run that code outside of your FastAPI app and make sure it works.

If you need to understand better how to use SQLAlchemy, check the tutorial: https://docs.sqlalchemy.org/en/13/orm/tutorial.html

Then, after you have your data, you have to convert it to have the shape of the model that you want to return. So, probably iterating in each of the rows in the result and putting all that in dicts, or something similar.

Then with that you can finally try to put it in a Pydantic model and see where the error is.

But first you have to debug the initial part and make sure that you are getting the data that you need, and then that you are converting it to the shape that you need.

[wedwardbeck]

As tiangolo said, verify you are returning the data you want (you should be). Your query should already be returning the data you need via SQLAlchemy here:

def get_multi_by_owner(
        self, db_session: Session, *, owner_id: str, skip=0, limit=100
    ) -> List[ShiftDetail]:
        return (
            db_session.query(self.model)
            .filter(ShiftDetail.owner_id == owner_id)
            .offset(skip)
            .limit(limit)
            .all()
        )

then add to your schema like so:

# Additional properties to return via API
class ShiftDetail(ShiftDetailBaseInDB):
    create_time: datetime
    owner: User
    
    class Config:
        orm_mode = True

and then be sure to import the User Schema from where you defined that (.user for example).
This will return the data as shown in paul121's comments:

[
  {
    "shift_id": "5240b998-7919-11ea-8f22-001a7dda7111",
    "owner_id": 56277c6c-7918-11ea-850b-001a7dda7111,
    "shift_date": "2020-04-08",
    "id": "87845423-795f-11ea-a51f-001a7dda7111",
    "create_time": "2020-04-08T13:09:39.067190",
    "owner": {
       "id": "..",
       "email": "admin@email.com",
       "hashed_password": "...",
    }
  }
]

It will be nested, and not inline like you expected - I didn't like how this returns but just means dealing with it differently when consuming.
To improve and only provide what you need, create an additional schema def in the users_schema.py (or what you have named it...) such as:

class UserShared(BaseModel):
    email: str

    class Config:
        orm_mode = True

import that in your Shiftdetail.py and use it in the schema in place of "User" :

# Additional properties to return via API
class ShiftDetail(ShiftDetailBaseInDB):
    create_time: datetime
    owner: UserShared
    
    class Config:
        orm_mode = True

Then it will just return the email, and not all the user fields.

[SurelyYouareJoking]

How to use Sqlalchemy session with databases

[Cozmo25]

As tiangolo said, verify you are returning the data you want (you should be). Your query should already be returning the data you need via SQLAlchemy here:

def get_multi_by_owner(
        self, db_session: Session, *, owner_id: str, skip=0, limit=100
    ) -> List[ShiftDetail]:
        return (
            db_session.query(self.model)
            .filter(ShiftDetail.owner_id == owner_id)
            .offset(skip)
            .limit(limit)
            .all()
        )

then add to your schema like so:

# Additional properties to return via API
class ShiftDetail(ShiftDetailBaseInDB):
    create_time: datetime
    owner: User
    
    class Config:
        orm_mode = True

and then be sure to import the User Schema from where you defined that (.user for example).
This will return the data as shown in paul121's comments:

[
  {
    "shift_id": "5240b998-7919-11ea-8f22-001a7dda7111",
    "owner_id": 56277c6c-7918-11ea-850b-001a7dda7111,
    "shift_date": "2020-04-08",
    "id": "87845423-795f-11ea-a51f-001a7dda7111",
    "create_time": "2020-04-08T13:09:39.067190",
    "owner": {
       "id": "..",
       "email": "admin@email.com",
       "hashed_password": "...",
    }
  }
]

It will be nested, and not inline like you expected - I didn't like how this returns but just means dealing with it differently when consuming.
To improve and only provide what you need, create an additional schema def in the users_schema.py (or what you have named it...) such as:

class UserShared(BaseModel):
    email: str

    class Config:
        orm_mode = True

import that in your Shiftdetail.py and use it in the schema in place of "User" :

# Additional properties to return via API
class ShiftDetail(ShiftDetailBaseInDB):
    create_time: datetime
    owner: UserShared
    
    class Config:
        orm_mode = True

Then it will just return the email, and not all the user fields.

Weird... I've done exactly what you say here for my model, but I'm still getting field required (type=value_error.missing) errors

class InputTemplateShared(InputTemplateBase):
    name: str

    class Config:
        orm_mode = True

# Shared Properties to return via API
class MetricShared(MetricBase):
    name: str

    class Config:
        orm_mode: True

Both imported into input_template_base.py schema file where I have:

# Properties shared by models stored in DB
class InputTemplateDetailInDBBase(InputTemplateDetailBase):
    template_id: Optional[int] = None
    metric_id: Optional[int] = None

    class Config:
        orm_mode = True

# Properties to return to client
class InputTemplateDetail(InputTemplateDetailInDBBase):
    template_name: InputTemplateShared
    metric_name: MetricShared

    class Config:
        orm_mode = True

Have checked my db query and it's returning the correct orm object and I can access those fields...

{
    "Template ID": 1,
    "Template Name": "anode_graphite",
    "Metric Name": "Thickness",
    "Metric Code": "w_ne",
    "Metric Value": 8.8e-05
}
{
    "Template ID": 1,
    "Template Name": "anode_graphite",
    "Metric Name": "Theoretical Capacity",
    "Metric Code": "Cth_ne",
    "Metric Value": 372.0
}

Any ideas?

[tiangolo]

@Cozmo25 I'm not sure I'm following correctly, but I think that by having something like:

# Properties to return to client
class InputTemplateDetail(InputTemplateDetailInDBBase):
    template_name: InputTemplateShared

    class Config:
        orm_mode = True

And then, before that:

class InputTemplateShared(InputTemplateBase):
    name: str

    class Config:
        orm_mode = True

It would expect something like:

{
  "template_name": {
    "name": "anode_graphite"
  }
}

Notice the nested dict for template_name.

Could that be the problem?

[Cozmo25]

@tiangolo Thank you for the response much appreciated.

It’s very likely the problem, I’m just not sure how to create the structure I need (not nested instead of nested) where I have a join between tables.