CRUD using ORM and Fastapi python using postman

1. Install python

Use below link for python installation

How to install Python in Window

2. Install Fastapi .

Use below link for install fastapi

How to install fastapi in python

3. Create main.py file in root directory

4. Create instance in main.py

from typing import Optional
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, SessionLocal, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

5. Create URL and methods (Create, Get, Update and Delete) in main.py

@app.get("/getposts")
async def get_posts(db: Session = Depends(get_db)):
    posts = db.query(models.Post).all()
    return {"Data" : posts}
#add post data
@app.post("/createpost", status_code=status.HTTP_201_CREATED)
async def create_post(new_post:Post, db: Session = Depends(get_db)):
    # postData = models.Post(title=new_post.title, content=new_post.content, published=new_post.published )
    postData = models.Post(**new_post.dict())
    db.add(postData)
    db.commit()
    db.refresh(postData)
    return {"Post": postData}

#get post by id
@app.get("/getpostbyID/{id}")
async def getpostbyID(id: int, db: Session = Depends(get_db)):
    get_post = db.query(models.Post).filter(models.Post.id == id).first() 

    if not get_post:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Post with id {id} not found")
    return {"Data": get_post}

#delete post
@app.delete("/deletepost/{id}")
async def deletePost(id: int, db: Session = Depends(get_db)):
    del_post = db.query(models.Post).filter(models.Post.id == id)

    if del_post.first() == None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"Post with id {id} does not exist")
    del_post.delete(synchronize_session=False)
    db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

#update post
@app.put("/updatepost/{id}")
async def updatepost(id: int, new_post: Post, db: Session = Depends(get_db)):
    query = db.query(models.Post).filter(models.Post.id == id)
    post = query.first()

    if post == None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"post with id {id} does not exist")
    query.update(new_post.dict(), synchronize_session=False)
    db.commit()
    return {"Data": query.first()}

6. Create database.py file for database connection

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


SQLALCHEMY_DATABASE_URL = 'postgresql://postgres:123456789@localhost/fastapi_ORM_DB'

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

7. Create model.py for table schema

from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.sql.expression import null
from .database import Base

class Post(Base):
    __tablename__ = "post"

    id = Column(Integer, primary_key=True, nullable=False)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    published = Column(Boolean, default=True)

Complete Code.

main.py

from typing import Optional
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, SessionLocal, get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

class Post(BaseModel):
    title:str
    content:str
    published : bool =True

@app.get("/test")
async def get_method():
    return {"MSG": "Get Method"}

@app.get("/sqlalchemy")
async def test_post(db: Session = Depends(get_db)):
    return {"MESSAGE": "test"}

@app.get("/getposts")
async def get_posts(db: Session = Depends(get_db)):
    posts = db.query(models.Post).all()
    return {"Data" : posts}
#add post data
@app.post("/createpost", status_code=status.HTTP_201_CREATED)
async def create_post(new_post:Post, db: Session = Depends(get_db)):
    postData = models.Post(**new_post.dict())
    db.add(postData)
    db.commit()
    db.refresh(postData)
    return {"Post": postData}

#get post by id
@app.get("/getpostbyID/{id}")
async def getpostbyID(id: int, db: Session = Depends(get_db)):
    get_post = db.query(models.Post).filter(models.Post.id == id).first() 

    if not get_post:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Post with id {id} not found")
    return {"Data": get_post}

#delete post
@app.delete("/deletepost/{id}")
async def deletePost(id: int, db: Session = Depends(get_db)):
    del_post = db.query(models.Post).filter(models.Post.id == id)

    if del_post.first() == None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f"Post with id {id} does not exist")
    del_post.delete(synchronize_session=False)
    db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

#update post
@app.put("/updatepost/{id}")
async def updatepost(id: int, new_post: Post, db: Session = Depends(get_db)):
    query = db.query(models.Post).filter(models.Post.id == id)
    post = query.first()

    if post == None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"post with id {id} does not exist")
    query.update(new_post.dict(), synchronize_session=False)
    db.commit()
    return {"Data": query.first()}

   

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker


SQLALCHEMY_DATABASE_URL = 'postgresql://postgres:123456789@localhost/fastapi_ORM_DB'

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

models.py

from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.sql.expression import null
from .database import Base

class Post(Base):
    __tablename__ = "post"

    id = Column(Integer, primary_key=True, nullable=False)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    published = Column(Boolean, default=True)

Run application

uvicorn main:app --reload

Happy learning 🙂