enter image description here file structure
api.py file
from typing import Optional
from typing import Union
from fastapi import FastAPI, Response, status, HTTPException, Depends
from fastapi.params import Body
from pydantic import BaseModel
from random import randrange
import psycopg2
from psycopg2.extras import RealDictCursor
import time
from sqlalchemy.orm import Session
from . import models
from .database import engine, get_db
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
class Post(BaseModel):
title: str
content: str
published: bool = True
while True:
try:
conn = psycopg2.connect(host='localhost', database='fastapi', user='postgres', password='legend100', cursor_factory=RealDictCursor)
cursor = conn.cursor()
print("Database connection was successfull !")
break
except Exception as error:
print("Connection to Database failed")
print("Error: ", error)
time.sleep(2)
my_posts = [{"title": "title of post 1", "content": "content of post 1", "id": 1}, {
"title": "favourite foods", "content": "I like pizza", "id": 2
}]
def find_post(id):
for p in my_posts:
if p["id"] == id:
return p
def find_index_post(id):
for i, p in enumerate(my_posts):
if p['id'] == id:
return i
@app.get("/")
async def read_root():
return {"message": "Welcome to my server"}
@app.get("/sqlalchemy")
def test_posts(db: Session = Depends(get_db)):
posts = db.query(models.Post).all()
return {"data": "posts"}
@app.get("/posts")
def get_posts():
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/latest")
def get_latest_post():
post = my_posts[len(my_posts)-1]
return {"detail": 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=f"post with id: {id} was not found")
# response.status_code = status.HTTP_404_NOT_FOUND
# return{"message": f"post with id: {id} was not found"}
return{"post_detail": post}
@app.delete("/posts/{id}")
def delete_post(id: int, status_code=status.HTTP_204_NO_CONTENT):
cursor.execute("""Delete from posts Where id = %s returning * """, (str(id),))
deleted_post = cursor.fetchone()
conn.commit()
#delete posts
#find index in the array that has required id
#my_posts.pop(index)
index = find_index_post(id)
if deleted_post == None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"post with id: {id} does not exist")
return Response(status_code=status.HTTP_204_NO_CONTENT)
@app.put("/posts/{id}")
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"post with id: {id} does not exist")
return {"data": updated_post}
database.py file
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# SQLALCHEMY_DATABASE_URL = 'postgresql://<username>:<password>@<ip-address/hostname>/<database_name>'
SQLALCHEMY_DATABASE_URL = "postgresql://postgres:legend100@localhost/fastapi"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autofulsh=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close
models.py file
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.sql.expression import text
from .database import Base
from sqlalchemy.sql.sqltypes import TIMESTAMP
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True, nullable=False)
title = Column(String, nullable=False)
content = Column(String, nullable=False)
publsihed = Column(Boolean, server_default='TRUE', nullable=False)
created_at = Column(TIMESTAMP(timezone=True), nullable=False, server_default=text('now()'))
i hit the run to start the api but i am geting this: Traceback (most recent call last): File "c:\Users\abhinav bhardwaj\Desktop\web project\app\api.py", line 11, in from . import models ImportError: attempted relative import with no known parent package
