0

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()
rfkortekaas
  • 6,049
  • 2
  • 27
  • 34
Kanav Raina
  • 43
  • 1
  • 9

0 Answers0