151 lines
5.6 KiB
Python
151 lines
5.6 KiB
Python
from fastapi import FastAPI, Depends, HTTPException, status, Query, Path, UploadFile, File, Form
|
|
from fastapi.staticfiles import StaticFiles
|
|
from sqlalchemy.orm import Session
|
|
import crud
|
|
import models
|
|
import schemas
|
|
from database import engine, get_db
|
|
from typing import List, Optional
|
|
import uvicorn
|
|
from utils import save_image, delete_image
|
|
import json
|
|
|
|
# Создание таблиц в БД
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI(title="AutoBro API", description="API для управления базой данных автомобилей")
|
|
|
|
# Добавляем обработку статических файлов
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
@app.get("/")
|
|
def read_root():
|
|
return {"message": "AutoBro API"}
|
|
|
|
@app.get("/cars", response_model=schemas.CarsResponse)
|
|
def get_cars(
|
|
skip: int = Query(0, description="Количество пропускаемых записей"),
|
|
limit: int = Query(100, description="Максимальное количество записей"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
cars = crud.get_cars(db, skip=skip, limit=limit)
|
|
total = crud.get_cars_count(db)
|
|
return {"cars": cars, "total": total}
|
|
|
|
@app.get("/cars/{car_id}", response_model=schemas.CarResponse)
|
|
def get_car(
|
|
car_id: int = Path(..., description="ID автомобиля", gt=0),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
car = crud.get_car(db, car_id=car_id)
|
|
if car is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Автомобиль не найден"
|
|
)
|
|
return {"car": car}
|
|
|
|
@app.post("/cars", response_model=schemas.CarResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_car(
|
|
car_data: str = Form(..., description="Данные автомобиля в JSON формате"),
|
|
image: UploadFile = File(None, description="Изображение автомобиля"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
try:
|
|
# Преобразуем строку JSON в словарь
|
|
car_dict = json.loads(car_data)
|
|
|
|
# Загружаем изображение, если оно предоставлено
|
|
if image:
|
|
image_path = await save_image(image)
|
|
car_dict["image"] = image_path
|
|
|
|
# Создаем объект Pydantic для валидации данных
|
|
car = schemas.CarCreate(**car_dict)
|
|
|
|
# Создаем запись в БД
|
|
db_car = crud.create_car(db=db, car=car)
|
|
return {"car": db_car}
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Неверный формат JSON"
|
|
)
|
|
|
|
@app.put("/cars/{car_id}", response_model=schemas.CarResponse)
|
|
async def update_car(
|
|
car_id: int = Path(..., description="ID автомобиля", gt=0),
|
|
car_data: str = Form(None, description="Данные автомобиля в JSON формате"),
|
|
image: UploadFile = File(None, description="Изображение автомобиля"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
# Проверяем существование автомобиля
|
|
existing_car = crud.get_car(db, car_id=car_id)
|
|
if existing_car is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Автомобиль не найден"
|
|
)
|
|
|
|
try:
|
|
# Преобразуем строку JSON в словарь, если она предоставлена
|
|
car_dict = {}
|
|
if car_data:
|
|
car_dict = json.loads(car_data)
|
|
|
|
# Загружаем новое изображение, если оно предоставлено
|
|
if image:
|
|
# Удаляем старое изображение, если есть
|
|
if existing_car.image:
|
|
delete_image(existing_car.image)
|
|
|
|
# Сохраняем новое изображение
|
|
image_path = await save_image(image)
|
|
car_dict["image"] = image_path
|
|
|
|
# Создаем объект Pydantic для валидации данных
|
|
car_update = schemas.CarUpdate(**car_dict)
|
|
|
|
# Обновляем запись в БД
|
|
updated_car = crud.update_car(db=db, car_id=car_id, car_update=car_update)
|
|
return {"car": updated_car}
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=str(e)
|
|
)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Неверный формат JSON"
|
|
)
|
|
|
|
@app.delete("/cars/{car_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_car(
|
|
car_id: int = Path(..., description="ID автомобиля", gt=0),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
# Получаем автомобиль перед удалением
|
|
car = crud.get_car(db, car_id=car_id)
|
|
if car is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Автомобиль не найден"
|
|
)
|
|
|
|
# Удаляем изображение, если есть
|
|
if car.image:
|
|
delete_image(car.image)
|
|
|
|
# Удаляем запись из БД
|
|
crud.delete_car(db=db, car_id=car_id)
|
|
return None
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run(app, host="0.0.0.0", port=3000)
|