362 lines
15 KiB
Python
362 lines
15 KiB
Python
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
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
# Создание таблиц в БД
|
||
models.Base.metadata.create_all(bind=engine)
|
||
|
||
app = FastAPI(title="AutoBro API", description="API для управления базой данных автомобилей")
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*",]
|
||
)
|
||
|
||
# Добавляем обработку статических файлов
|
||
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_super_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.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
|
||
|
||
# Эндпоинт для проверки текущего пользователя (только для отладки)
|
||
@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)
|
||
|
||
# Персонал ---------
|
||
|
||
@app.get("/personal", response_model=schemas.PersonalListResponse)
|
||
def get_all_personal(
|
||
skip: int = Query(0, description="Количество пропускаемых записей"),
|
||
limit: int = Query(100, description="Максимальное количество записей"),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
staff = crud.get_all_personal(db, skip=skip, limit=limit)
|
||
total = crud.get_personal_count(db)
|
||
return {"staff": staff, "total": total}
|
||
|
||
@app.get("/personal/{personal_id}", response_model=schemas.PersonalResponse)
|
||
def get_personal(
|
||
personal_id: int = Path(..., description="ID сотрудника", gt=0),
|
||
db: Session = Depends(get_db)
|
||
):
|
||
personal = crud.get_personal(db, personal_id=personal_id)
|
||
if personal is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Сотрудник не найден"
|
||
)
|
||
return {"personal": personal}
|
||
|
||
@app.post("/personal", response_model=schemas.PersonalResponse, status_code=status.HTTP_201_CREATED)
|
||
async def create_personal(
|
||
personal_data: str = Form(..., description="Данные сотрудника в JSON формате"),
|
||
photo: UploadFile = File(None, description="Фотография сотрудника"),
|
||
db: Session = Depends(get_db),
|
||
current_admin: models.Admin = Depends(auth.get_current_admin)
|
||
):
|
||
try:
|
||
# Преобразуем строку JSON в словарь
|
||
personal_dict = json.loads(personal_data)
|
||
|
||
# Загружаем фото, если оно предоставлено
|
||
if photo:
|
||
photo_path = await save_image(photo)
|
||
personal_dict["photo"] = photo_path
|
||
|
||
# Создаем объект Pydantic для валидации данных
|
||
personal = schemas.PersonalCreate(**personal_dict)
|
||
|
||
# Создаем запись в БД
|
||
db_personal = crud.create_personal(db=db, personal=personal)
|
||
return {"personal": db_personal}
|
||
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("/personal/{personal_id}", response_model=schemas.PersonalResponse)
|
||
async def update_personal(
|
||
personal_id: int = Path(..., description="ID сотрудника", gt=0),
|
||
personal_data: str = Form(None, description="Данные сотрудника в JSON формате"),
|
||
photo: UploadFile = File(None, description="Фотография сотрудника"),
|
||
db: Session = Depends(get_db),
|
||
current_admin: models.Admin = Depends(auth.get_current_admin)
|
||
):
|
||
# Проверяем существование сотрудника
|
||
existing_personal = crud.get_personal(db, personal_id=personal_id)
|
||
if existing_personal is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Сотрудник не найден"
|
||
)
|
||
|
||
try:
|
||
# Преобразуем строку JSON в словарь, если она предоставлена
|
||
personal_dict = {}
|
||
if personal_data:
|
||
personal_dict = json.loads(personal_data)
|
||
|
||
# Загружаем новую фотографию, если она предоставлена
|
||
if photo:
|
||
# Удаляем старую фотографию, если есть
|
||
if existing_personal.photo:
|
||
delete_image(existing_personal.photo)
|
||
|
||
# Сохраняем новую фотографию
|
||
photo_path = await save_image(photo)
|
||
personal_dict["photo"] = photo_path
|
||
|
||
# Создаем объект Pydantic для валидации данных
|
||
personal_update = schemas.PersonalUpdate(**personal_dict)
|
||
|
||
# Обновляем запись в БД
|
||
updated_personal = crud.update_personal(db=db, personal_id=personal_id, personal_update=personal_update)
|
||
return {"personal": updated_personal}
|
||
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("/personal/{personal_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
def delete_personal(
|
||
personal_id: int = Path(..., description="ID сотрудника", gt=0),
|
||
db: Session = Depends(get_db),
|
||
current_admin: models.Admin = Depends(auth.get_current_admin)
|
||
):
|
||
# Получаем сотрудника перед удалением
|
||
personal = crud.get_personal(db, personal_id=personal_id)
|
||
if personal is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="Сотрудник не найден"
|
||
)
|
||
|
||
# Удаляем фотографию, если есть
|
||
if personal.photo:
|
||
delete_image(personal.photo)
|
||
|
||
# Удаляем запись из БД
|
||
crud.delete_personal(db=db, personal_id=personal_id)
|
||
return None
|