FastAPI SQLModel unable to filter with where

167 Views Asked by At

I have the following Code:

helpers/database.py

from sqlmodel import SQLModel, Session, create_engine

# Setup SQLite DB
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"
connect_args = {"check_same_thread": False}


engine = create_engine(sqlite_url, echo = True, connect_args=connect_args)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

models/creator.py

from typing import Optional
from sqlalchemy import String, Column
from sqlmodel import Field, SQLModel
from pydantic import EmailStr

class CreatorPost(SQLModel):
    first_name: str = Field(default="First")
    last_name: str = Field(default="Last")
    display: str = Field(default="Username", index=True)
    email: str = Field(sa_column=Column("email", String, index=True, unique=True))
    image: Optional[str] = Field(default="/img/UserImg/<display>.webp")
    avatar: Optional[str] = Field(default="/img/avatars/<display>.webp")

class Creator(CreatorPost, table=True):
#    __table_args__ = (UniqueConstraint("email"),)
    id: Optional[int] = Field(default=None,index=True, primary_key=True)
    flags: Optional[str] = Field(default=None)
    rating: Optional[int] = Field(default=100)
    disabled: Optional[bool] = Field(default=False)
    value: Optional[str] = Field(default=None)

class CreatorRead(CreatorPost):
    id: Optional[int] = Field(default=None,index=True, primary_key=True)
    flags: Optional[str] = Field(default=None)
    rating: Optional[int] = Field(default=100)
    disabled: Optional[bool] = Field(default=False)
    value: Optional[str] = Field(default=None)
    email: Optional[str]

routers/creators.py

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlmodel import Session, select
from typing import List
from sqlalchemy.exc import IntegrityError
from pydantic import EmailStr


from api.models.creator import Creator, CreatorPost, CreatorRead
from api.helpers.database import get_session


router = APIRouter(
    prefix="/creator",
    responses={404:{"Description": "Not Found"}}
)

@router.post(
    "/add",
    response_model=CreatorRead,
)
async def creator_post(
    *,
    session: Session = Depends(get_session),
    creator: CreatorPost
    ):
    try:
        db_creator = Creator.from_orm(creator)
        session.add(db_creator)
        session.commit()
        session.refresh(db_creator)
        return db_creator
    except IntegrityError as error:
        raise HTTPException(
            status_code=404,
            detail=f"{error.__dict__['orig']}"
            )

@router.get(
    "/all",
    response_model=List[CreatorRead],
)
async def creator_get(
    *,
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = Query(default=100, le=100),
    ):
    creator = session.exec(select(Creator).offset(offset).limit(limit)).all()
    return creator

@router.get(
    "/id/{id}",
    response_model=CreatorRead,
)
async def creator_by_id(
    id: int,
    session: Session = Depends(get_session),
    ):
    creator = session.get(Creator, id)
    if not creator:
        raise HTTPException(status_code=404)
    return creator

@router.get(
    "/display",
    response_model=CreatorRead,
)
async def creator_by_display(
    data: str,
    session: Session = Depends(get_session),
    ):
    print(f"data:\n \
    ########################################\n \
    {data}\n \
    ########################################\n \
    ")
    query = select(Creator).where(Creator.display == f"'{data}'")
    print(f"query:\n \
    ########################################\n \
    {query}\n \
    ########################################\n \
    ")
    creator = session.exec(query)
    if not creator:
        raise HTTPException(status_code=404)
    return creator

main.py

from typing import Optional
from fastapi import FastAPI
from sqlmodel import Field, Session, SQLModel, create_engine, select

from api.helpers.database import create_db_and_tables, get_session, engine

# Import Routers
from .routers import creators

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

app.include_router(creators.router)

I can successfully insert a record:

{
  "first_name": "John",
  "last_name": "Doe",
  "display": "JohnDoe",
  "email": "[email protected]",
  "image": "/img/UserImg/<display>.webp",
  "avatar": "/img/avatars/<display>.webp"
}

I can also successfully retrieve all records and by id, but when I try to access /creator/display?data=JohnDoe I get the following back:

{
  "first_name": "First",
  "last_name": "Last",
  "display": "Username",
  "email": null,
  "image": "/img/UserImg/<display>.webp",
  "avatar": "/img/avatars/<display>.webp",
  "id": null,
  "flags": null,
  "rating": 100,
  "disabled": false,
  "value": null
}

which is not correct, and the value of query is:

SELECT creator.email, creator.first_name, creator.last_name, creator.display, creator.image, creator.avatar, creator.id, creator.flags, creator.rating, creator.disabled, creator.value 
FROM creator 
WHERE creator.display = :display_1

which I am expecting to be:

SELECT creator.email, creator.first_name, creator.last_name, creator.display, creator.image, creator.avatar, creator.id, creator.flags, creator.rating, creator.disabled, creator.value 
FROM creator 
WHERE creator.display = 'JohnDoe'

I'm sure I'm just missing something stupid, but I'm pulling my hair out trying to figure this out.

0

There are 0 best solutions below