CRUD via SQLAlchemy

Get All

• We will have to import the database.py file and the models.py file

from . import models
from app.database import engine, get_db

• Next we will create the function with that calls the get_db dependency and creates the query

@app.get("/sqlalchemy")
# We are saving the dependency and session in a variable (db)
def test_post(db: Session = Depends(get_db)):
    
    # We call that variable
    # Set to query
    # Add the class name from the models.py Table
    # Fetch all rows with .all()
    posts = db.query(models.Post).all()

    return { "data": posts }

Note: Syntax for all queries is:

• db variable

• query to interrogate

• models.<CLASS_NAME> - table that we want to use

• .add() - Fetches all rows

Get by ID

• We will fetch the post by ID by using the query and filtering for the ID in the models that is equal to the ID passed in

def get_post(id: int, db: Session = Depends(get_db)):

    post = db.query(models.Post).filter(models.Post.id == id).first()


    if not post:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"post with id {id} was not found")

    return {"post_details": post}

Note: If you know that you only have a single post with said ID, use the method .first() instead of .all()

• This is more efficient as .all() will look further through all posts, even tho it has found the post with the ID specified

• Meaning that it uses resources that it doesn't need

Create

• We're basically doing the same thing as RAW SQL but we are passing in the variables directly

@app.post("/posts", status_code=status.HTTP_201_CREATED)
def create_posts(post: Post, db: Session = Depends(get_db)):

    # Map table fields to data to add
    new_post = models.Post(title=post.title, content=post.content, published=post.published)

    # Add post to database
    db.add(new_post)

    # Commit it to the database
    db.commit()

    # Returning - Basically refresh table and retrieve the post in order to get the
    db.refresh(new_post)

    return { "data": new_post }

Delete

• The first part of the query stays the same as the Get ID one

@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int, db: Session = Depends(get_db))

    post = db.query(models.Post).filter(models.Post.id == id)



    if post.first() == None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"post with {id} does not exist")

    post.delete(synchronize_session=False)
    db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

• We do call the .first() method to check if there is no ID

• If there isn't it will raise an exception

• Outside the IF statement we will delete the post with the ID

• Then we have to commit the changes to the database

Update

@app.put("/posts/{id}")
def update_posts(id: int, post_schema: Post, db: Session = Depends(get_db)):

    post_query = db.query(models.Post).filter(models.Post.id == id)

    post = post_query.first()

    # Raising a 404 if post index position of post is not found
    if post == None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"post with {id} does not exist")

    post_query.update(post_schema.dict(), synchronize_session=False)

    db.commit()

    return {"data": post_query.first() }  # Return new post post

• We are saving the query in the post_query variable

• Then we are getting the first post

• If that post does not exist, it raises an error

• If it does it updates the database entry with the post_schemathat already is passed in

• Commits this to the database

• To return it to the user, we just run another query