1

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.

1
  • also i think you could change this : query = select(Creator).where(Creator.display == f"'{data}'") to query = select(Creator).where(Creator.display == data) Commented Nov 18, 2023 at 18:24

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.