main.py - master - Zero1l/aaa

main.py
106 строк3 KB
Zero1l
update main.py
03 фев 2026, 14:51
03 фев 2026, 14:51a09ca65
# aaa
# aaa
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List, Optional

from .database import SessionLocal, engine
from .models import Advertisement
from .schemas import (
    AdvertisementCreate,
    AdvertisementUpdate,
    AdvertisementResponse,
)
from .database import Base
from .routers import user as user_routes
from .routers import advertisement as ad_routes


app = FastAPI(title="Advertisements service with auth (SQLAlchemy)")




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


@app.post("/advertisement", response_model=AdvertisementResponse)
def create_advertisement(
    advertisement: AdvertisementCreate,
    db: Session = Depends(get_db),
):
    ad = Advertisement(**advertisement.dict())
    db.add(ad)
    db.commit()
    db.refresh(ad)
    return ad


@app.get(
    "/advertisement/{advertisement_id}",
    response_model=AdvertisementResponse,
)
def get_advertisement(advertisement_id: int, db: Session = Depends(get_db)):
    ad = db.get(Advertisement, advertisement_id)
    if ad is None:
        raise HTTPException(status_code=404, detail="Advertisement not found")
    return ad


@app.patch(
    "/advertisement/{advertisement_id}",
    response_model=AdvertisementResponse,
)
def update_advertisement(
    advertisement_id: int,
    advertisement_update: AdvertisementUpdate,
    db: Session = Depends(get_db),
):
    ad = db.get(Advertisement, advertisement_id)
    if ad is None:
        raise HTTPException(status_code=404, detail="Advertisement not found")

    for field, value in advertisement_update.dict(exclude_unset=True).items():
        setattr(ad, field, value)

    db.commit()
    db.refresh(ad)
    return ad


@app.delete("/advertisement/{advertisement_id}")
def delete_advertisement(advertisement_id: int, db: Session = Depends(get_db)):
    ad = db.get(Advertisement, advertisement_id)
    if ad is None:
        raise HTTPException(status_code=404, detail="Advertisement not found")

    db.delete(ad)
    db.commit()
    return {"status": "deleted"}


@app.get("/advertisement", response_model=List[AdvertisementResponse])
def search_advertisements(
    title: Optional[str] = None,
    author: Optional[str] = None,
    min_price: Optional[int] = None,
    max_price: Optional[int] = None,
    db: Session = Depends(get_db),
):
    query = db.query(Advertisement)

    if title:
        query = query.filter(Advertisement.title.contains(title))
    if author:
        query = query.filter(Advertisement.author.contains(author))
    if min_price is not None:
        query = query.filter(Advertisement.price >= min_price)
    if max_price is not None:
        query = query.filter(Advertisement.price <= max_price)

    return query.all()