With the code below and following sequence I suspect an IntegrityError with a Foreign Key violation. Unfortunately the error isn't raised.
- Create
Customer(id=1)
- Create
Booking(cid=1)
- Delete
Customer(id=1)
#main.py
from typing import List
from fastapi.params import Depends
from schemas import Customer, ShowCustomer, Booking, ShowBooking
from fastapi import FastAPI, HTTPException
from database import get_db, engine, Base
from sqlalchemy.orm import Session
import models
models.Base.metadata.create_all(engine)
app = FastAPI()
@app.post("/customer")
async def customer(req: Customer, db: Session=Depends(get_db)):
new_customer=models.Customer(name=req.name, type=req.type)
db.add(new_customer)
db.commit()
db.refresh(new_customer)
return new_customer
@app.get("/customer", response_model=List[ShowCustomer])
async def create(db: Session=Depends(get_db)):
customers=db.query(models.Customer).all()
return customers
@app.delete('/customer/{id}')
def destory(id, db: Session=Depends(get_db)):
customer=db.query(models.Customer).filter(models.Customer.id == id)
if not customer.first():
raise HTTPException(status_code=404,
detail=f'Blog with the id {id} is not available '
)
customer.delete(synchronize_session=False)
db.commit()
return 'done'
@app.post("/booking")
async def read_root(req: Booking, db: Session=Depends(get_db)):
new_booking=models.Booking(name=req.name, type=req.type, cid=1)
db.add(new_booking)
db.commit()
db.refresh(new_booking)
return new_booking
@app.get("/booking",response_model=List[ShowBooking])
async def read_root(db: Session=Depends(get_db)):
bookings=db.query(models.Booking).all()
return bookings
# schemas.py
from pydantic import BaseModel
class Customer(BaseModel):
name:str
type:str
class ShowCustomer(Customer):
id:int
class Config():
orm_mode = True
class Booking(BaseModel):
name:str
type:str
class ShowBooking(Booking):
id:int
cid:int
class Config():
orm_mode = True
#models.py
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from database import Base
class Customer(Base):
__tablename__ = "customers"
id = Column(Integer, primary_key=True, index=True)
name = Column(String)
type = Column(String)
booking = relationship("Booking", back_populates="customer")
class Booking(Base):
__tablename__ = "bookings"
id = Column(Integer, primary_key=True, index=True)
cid = Column(Integer, ForeignKey("customers.id"), nullable=False)
name = Column(String)
type = Column(String)
customer = relationship('Customer', back_populates="booking")
#database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
SQLALCHEMY_DATABASE_URL = 'sqlite:///./cloud2.db'
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()
def get_db():
db=SessionLocal()
try:
yield db
finally:
db.close()