FastAPI로 Postman CRUD

Postman을 통해서 CRUD 해보기
https://web.postman.com

 

1. 전체 코드

 

database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
	SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

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

---

models.py

from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Test(Base):
    __tablename__ = "test1"  # 테이블 이름 정의

    # 테이블의 열을 정의
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String)
    confirm = Column(Boolean, default=True)

---

main.py

from fastapi import FastAPI, Depends, HTTPException
import models
from database import engine, SessionLocal
from sqlalchemy.orm import Session
from pydantic import BaseModel

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

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

class TestConfirm(BaseModel):
    title: str
    description: str
    confirm: bool

@app.get("/")
async def read_all(db: Session = Depends(get_db)):
    return db.query(models.Test).all()

@app.get("/{test_id}")
async def read_test1(test_id: int, db: Session = Depends(get_db)):
    test_model = db.query(models.Test).filter(models.Test.id == test_id).first()
    if test_model is not None:
        return test_model
    raise http_exception()

@app.post("/")
async def create_test1(test: TestConfirm, db: Session = Depends(get_db)):

    test_model = models.Test()
    test_model.title = test.title
    test_model.description = test.description
    test_model.confirm = test.confirm

    db.add(test_model)
    db.flush()
    db.commit()
    return successful_response(200)

@app.put("/{test_id}")
async def update_test1(test_id: int, test: TestConfirm, db: Session = Depends(get_db)):

    test_model = db.query(models.Test).filter(models.Test.id == test_id).first()

    if test_model is None:
        raise http_exception()

    test_model.title = test.title
    test_model.description = test.description
    test_model.confirm = test.confirm

    db.add(test_model)
    db.flush()
    db.commit()
    return successful_response(200)

@app.delete("/{test_id}")
async def delete_test1(test_id: int, db: Session = Depends(get_db)):

    test_model = db.query(models.Test).filter(models.Test.id == test_id).first()

    if test_model is None:
        raise http_exception()

    db.query(models.Test).filter(models.Test.id == test_id).delete()
    db.flush()
    db.commit()
    return successful_response(200)


def successful_response(status_code:int):
    return {
        'status': status_code,
        'transaction': 'Successful'
    }

def http_exception():
    return HTTPException(status_code=404, detail="Test not found")

 

2. 데이터 만들기(POST 요청)

 

3. 데이터 전체 조회하기(GET 요청)

테이블의 전체 데이터가 조회된다.

 

4. 특정 데이터 조회하기(GET 요청)

지정한 id의 값으로 데이터를 조회한다.

왼쪽: id가 1인 데이터 / 오른쪽: id가 2인 데이터

 

5. 데이터 업데이트하기(PUT 요청)

id가 2번인 데이터가 업데이트 된다.

컬럼 타입이나 하나의 컬럼을 제외 했을 경우

 

6. 데이터 삭제하기(DELETE 요청)

id가 1번인 데이터가 삭제 된다.