CODEWITHSUNDEEP
pythonsqlalchemyfastapipydantic
Michael Knoth
Michael Knoth

Reputation: 78

Mapping issues from Sqlalchemy to Pydantic - from_orm failed

I'm trying to get my result dictonary from sqlalchemy automatically to the Pydantic output for Fastapi to maps using the from_orm method, but I always get a validation error.

File "pydantic\main.py", line 508, in pydantic.main.BaseModel.from_orm pydantic.error_wrappers.ValidationError: 2 validation errors for Category name field required (type=value_error.missing) id field required (type=value_error.missing)

If I create the objects with the Pydantic schema myself and add them to the list, the method works. What would I have to change for from_orm to work? Did I possibly miss something in the documentation? https://pydantic-docs.helpmanual.io/usage/models/#orm-mode-aka-arbitrary-class-instances https://fastapi.tiangolo.com/tutorial/sql-databases/#use-pydantics-orm_mode

or is there another/better way to turn the ResultProxy into a Pydantic capable output?

The output I get from the database method is the following:

[{'id': 1, 'name': 'games', 'parentid': None}, {'id': 2, 'name': 'computer', 'parentid': None}, {'id': 3, 'name': 'household', 'parentid': None}, {'id': 10, 'name': 'test', 'parentid': None}]]

Models.py

from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, Integer, Numeric, String, Text, text, Table
from sqlalchemy.orm import relationship, mapper
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
metadata = Base.metadata

category = Table('category', metadata,
                 Column('id', Integer, primary_key=True),
                 Column('name', String(200)),
                 Column('parentid', Integer),
                 )


class Category(object):
    def __init__(self, cat_id, name, parentid):
        self.id = cat_id
        self.name = name
        self.parentid = parentid


mapper(Category, category)

Schemas.py

from pydantic import BaseModel, Field

class Category(BaseModel):
    name: str
    parentid: int = None
    id: int
    class Config:
        orm_mode = True

main.py

def result_proxy_to_Dict(results: ResultProxy):
    d, a = {}, []
    for rowproxy in results:
        # rowproxy.items() returns an array like [(key0, value0), (key1, value1)]
        for column, value in rowproxy.items():
            # build up the dictionary
            d = {**d, **{column: value}}
        a.append(d)
    return a

def crud_read_cat(db: Session) -> dict:
    # records = db.query(models.Category).all()
    #query = db.query(models.Category).filter(models.Category.parentid == None)
    s = select([models.Category]). \
        where(models.Category.parentid == None)

    result = db.execute(s)
    #print(type(result))

    #print(result_proxy_to_Dict(result))
    #results = db.execute(query)
    # result_set = db.execute("SELECT id, name, parentid FROM public.category;")

    # rint(type(result_set))
    # for r in result_set:
    #    print(r)
    # return [{column: value for column, value in rowproxy.items()} for rowproxy in result_set]
    # return await databasehelper.database.fetch_all(query)
    return result_proxy_to_Dict(result)
    #return results


@router.get("/category/", response_model=List[schemas.Category], tags=["category"])
async def read_all_category(db: Session = Depends(get_db)):
    categories = crud_read_cat(db)
    context = []
    print(categories)
    co_model = schemas.Category.from_orm(categories)
    # print(co_model)
    for row in categories:
        print(row)
        print(row.get("id", None))
        print(row.get("name", None))
        print(row.get("parentid", None))
        tempcat = schemas.Category(id=row.get("id", None), name=row.get("name", None),
                                  parentid=row.get("parentid", None))
        context.append(tempcat)
    #for dic in [dict(r) for r in categories]:
        # print(dic)
        # print(dic.get("category_id", None))
        # print(dic.get("category_name", None))
        # print(dic.get("category_parentid", None))
    #    tempcat = schemas.Category(id=dic.get("category_id", None), name=dic.get("category_name", None),
    #                               parentid=dic.get("category_parentid", None))
    #    context.append(tempcat)

    return context

Upvotes: 3

Views: 21620

Answers (2)

Brendan Murphy
Brendan Murphy

Reputation: 46

New to this my self so cant promise the best answer but I noticed if you simply put Optional in the schema it works.

`

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

        class Config:
            orm_mode = True
`

Still returning that info in the response:

[
  {
    "name": "games",
    "parentid": null,
    "id": 1
  },
  {
    "name": "computer",
    "parentid": null,
    "id": 2
  },
  {
    "name": "household",
    "parentid": null,
    "id": 3
  },
  {
    "name": "test",
    "parentid": null,
    "id": 10
  }
]

Likely still some sort of validation error but seems like a usable work around for now.

Upvotes: 1

Barnez299
Barnez299

Reputation: 67

I just had the same problem. I think its related to pydantic nonethless. Please have a look at this link for more information https://github.com/samuelcolvin/pydantic/issues/506.

But having changed my model:

class Student(BaseModel):
    id: Optional[int] --- changed to optional
    name: Optional [str]
    surname: Optional [str]
    email: Optional [str]

The error validation goes away. Its a funny error - given that the entries in my database still updated with the values...I am new to fastAPI also so the workaround and the error does not really make sense for now....but yes it worked. Thank you

Upvotes: 0

Related Questions