56 lines
No EOL
2.2 KiB
Python
56 lines
No EOL
2.2 KiB
Python
from fastapi import HTTPException
|
|
from typing import List
|
|
from models.subscription import Subscription
|
|
from fastapi import Depends
|
|
from database import SessionLocal
|
|
from fastapi import FastAPI
|
|
|
|
from schemas.subscription import SubscriptionCreate, Subscription
|
|
app = FastAPI()
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
@app.post("/newsletter/", response_model=Subscription)
|
|
def sub_newsletter(newsletter: SubscriptionCreate, db = Depends(get_db)):
|
|
db_newsletter = Subscription(**newsletter.dict())
|
|
db.add(db_newsletter)
|
|
db.commit()
|
|
db.refresh(db_newsletter)
|
|
return db_newsletter
|
|
|
|
@app.get("/subscriptions/", response_model=List[Subscription])
|
|
def read_subscriptions(skip: int = 0, limit: int = 100, db = Depends(get_db)):
|
|
subscriptions = db.query(Subscription).offset(skip).limit(limit).all()
|
|
return subscriptions
|
|
|
|
@app.get("/subscriptions/{subscription_id}", response_model=Subscription)
|
|
def read_subscription(subscription_id: int, db = Depends(get_db)):
|
|
db_subscription = db.query(Subscription).filter(Subscription.id == subscription_id).first()
|
|
if db_subscription is None:
|
|
raise HTTPException(status_code=404, detail="subscription not found")
|
|
return db_subscription
|
|
|
|
@app.put("/subscriptions/{subscription_id}", response_model=Subscription)
|
|
def update_subscription(subscription_id: int, subscription: SubscriptionCreate, db = Depends(get_db)):
|
|
db_subscription = db.query(Subscription).filter(Subscription.id == subscription_id).first()
|
|
if db_subscription is None:
|
|
raise HTTPException(status_code=404, detail="subscription not found")
|
|
for key, value in subscription.model_dump().items():
|
|
setattr(db_subscription, key, value)
|
|
db.commit()
|
|
db.refresh(db_subscription)
|
|
return db_subscription
|
|
|
|
@app.delete("/subscriptions/{subscription_id}")
|
|
def delete_subscription(subscription_id: int, db = Depends(get_db)):
|
|
db_subscription = db.query(Subscription).filter(Subscription.id == subscription_id).first()
|
|
if db_subscription is None:
|
|
raise HTTPException(status_code=404, detail="subscription not found")
|
|
db.delete(db_subscription)
|
|
db.commit()
|
|
return {"message": "subscription deleted successfully"} |