from fastapi import FastAPI, Depends, HTTPException, status, Query, Path, UploadFile, File, Form from fastapi.staticfiles import StaticFiles from fastapi.security import OAuth2PasswordRequestForm 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 from datetime import timedelta import auth # Создание таблиц в БД models.Base.metadata.create_all(bind=engine) app = FastAPI(title="AutoBro API", description="API для управления базой данных автомобилей") # Добавляем обработку статических файлов app.mount("/static", StaticFiles(directory="static"), name="static") # Эндпоинты для авторизации @app.post("/token", response_model=schemas.Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)): admin = auth.authenticate_admin(db, form_data.username, form_data.password) if not admin: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Неверное имя пользователя или пароль", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES) access_token = auth.create_access_token( data={"sub": admin.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.post("/admins", response_model=schemas.Admin, status_code=status.HTTP_201_CREATED) def create_admin(admin: schemas.AdminCreate, db: Session = Depends(get_db), current_admin: models.Admin = Depends(auth.get_current_admin)): # Проверка, что создать админа может только существующий админ db_admin = crud.get_admin_by_username(db, username=admin.username) if db_admin: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Пользователь с таким именем уже существует" ) return crud.create_admin(db=db, admin=admin) @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), current_admin: models.Admin = Depends(auth.get_current_admin) ): 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), current_admin: models.Admin = Depends(auth.get_current_admin) ): # Проверяем существование автомобиля 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), current_admin: models.Admin = Depends(auth.get_current_admin) ): # Получаем автомобиль перед удалением 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 # Эндпоинт для проверки текущего пользователя (только для отладки) @app.get("/users/me", response_model=schemas.Admin) def read_users_me(current_admin: models.Admin = Depends(auth.get_current_admin)): return current_admin if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=3000)