Snips & Tips
Ask or search…
K

Fast API CRUD

from typing import Optional
from fastapi import Body, FastAPI, Response, status, HTTPException
from pydantic import BaseModel
from random import randrange
import psycopg2
from psycopg2.extras import RealDictCursor
import time
app = FastAPI()
class Post(BaseModel):
title: str
content: str
published: bool = True
rating: Optional[int] = None
while True:
try:
conn = psycopg2.connect(host="localhost", database="fastapi",
user="postgres", password="admin", cursor_factory=RealDictCursor)
cursor = conn.cursor()
print("Database connection successful")
break
except Exception as error:
print("Connection failed")
print("Error: ", error)
time.sleep(2)
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/posts")
def post():
cursor.execute("SELECT * from posts")
posts = cursor.fetchall()
return {"data": posts}
@app.post("/posts", status_code=status.HTTP_201_CREATED)
def create_posts(post: Post):
cursor.execute("""INSERT INTO posts (title, content, published) VALUES
(%s, %s, %s) RETURNING * """,
(post.title, post.content, post.published))
new_post = cursor.fetchone()
conn.commit()
return {"data": new_post}
@app.get("/posts/{id}")
def get_post(id: int):
cursor.execute("""SELECT * from posts WHERE id = %s""", (str(id)))
post = cursor.fetchone()
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Post not found")
return {
"post_details": post
}
@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(id: int):
cursor.execute("DELETE FROM posts WHERE id = %s RETURNING *", (str(id),))
deleted_post = cursor.fetchone()
conn.commit()
if deleted_post == None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=f"{id} not found")
return Response(status_code=status.HTTP_204_NO_CONTENT)
@app.put("/posts/{id}", status_code=status.HTTP_201_CREATED)
def update_post(id: int, post: Post):
cursor.execute("""UPDATE posts SET title = %s, content = %s, published = %s
WHERE id = %s RETURNING *""", (post.title, post.content,
post.published, str(id)))
updated_post = cursor.fetchone()
conn.commit()
if updated_post == None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail=f"{id} not found")
return {"data": updated_post}