main.py - master - runesmaker/bookbase

main.py
219 строк7 KB
pivnenkoanton0
feat: Базовая логика
14 июн 2026, 15:28
14 июн 2026, 15:28b99564b
from fastapi import FastAPI, HTTPException, status
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from pydantic import BaseModel
from typing import Optional, List

from database import get_db_connection, init_db

# 1. Инициализация приложения FastAPI
app = FastAPI(
    title="Моя библиотека API",
    description="Простой API для управления коллекцией",
    version="1.0.0"
)

# === НОВОЕ: Раздача статических файлов (HTML, JS, CSS) ===
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/", response_class=FileResponse)
def read_root():
    return "static/index.html"


# 2. Pydantic-модели
class BookCreate(BaseModel):
    title: str
    author: str
    isbn: Optional[str] = None
    publisher: Optional[str] = None
    publish_year: Optional[int] = None
    language: str = "Русский"
    pages: Optional[int] = None
    genre: Optional[str] = None
    series: Optional[str] = None
    series_number: Optional[int] = None
    binding: Optional[str] = None
    book_format: Optional[str] = None
    condition: Optional[str] = None
    location: Optional[str] = None
    status: str = "Планирую"
    rating: Optional[float] = None
    date_finished: Optional[str] = None
    notes: Optional[str] = None
    is_favorite: int = 0


# Модель для обновления: все поля необязательны (Optional)
class BookUpdate(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    isbn: Optional[str] = None
    publisher: Optional[str] = None
    publish_year: Optional[int] = None
    language: Optional[str] = None
    pages: Optional[int] = None
    genre: Optional[str] = None
    series: Optional[str] = None
    series_number: Optional[int] = None
    binding: Optional[str] = None
    book_format: Optional[str] = None
    condition: Optional[str] = None
    location: Optional[str] = None
    status: Optional[str] = None
    rating: Optional[float] = None
    date_finished: Optional[str] = None
    notes: Optional[str] = None
    is_favorite: Optional[int] = None


# 3. Событие запуска: создаем таблицу в БД, если её еще нет
@app.on_event("startup")
def startup_event():
    init_db()


# ==========================================
# CRUD ОПЕРАЦИИ
# ==========================================

# --- CREATE ---
@app.post("/books", status_code=status.HTTP_201_CREATED, summary="Добавить новую книгу")
def create_book(book: BookCreate):
    conn = get_db_connection()
    cursor = conn.cursor()

    query = """
    INSERT INTO books (
        title, author, isbn, publisher, publish_year, language, pages, 
        genre, series, series_number, binding, book_format, condition, 
        location, status, rating, date_finished, notes, is_favorite
    ) VALUES (
        :title, :author, :isbn, :publisher, :publish_year, :language, :pages, 
        :genre, :series, :series_number, :binding, :book_format, :condition, 
        :location, :status, :rating, :date_finished, :notes, :is_favorite
    ) RETURNING *
    """

    try:
        cursor.execute(query, book.model_dump())
        new_book = cursor.fetchone()
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=500, detail=f"Ошибка БД: {str(e)}")
    finally:
        conn.close()

    return dict(new_book)


# --- READ (ALL) ---
@app.get("/books", summary="Получить список всех книг")
def get_books(search: Optional[str] = None, status: Optional[str] = None, is_favorite: Optional[int] = None):
    """
    Возвращает список книг. Поддерживает простую фильтрацию:
    - search: поиск по названию или автору
    - status: фильтр по статусу (напр. "Читаю")
    - is_favorite: 1 для любимых, 0 для обычных
    """
    conn = get_db_connection()
    cursor = conn.cursor()

    # Базовый запрос
    query = "SELECT * FROM books WHERE 1=1"
    params = {}

    # Динамическое добавление условий фильтрации
    if search:
        query += " AND (title LIKE :search OR author LIKE :search)"
        params["search"] = f"%{search}%"
    if status:
        query += " AND status = :status"
        params["status"] = status
    if is_favorite is not None:
        query += " AND is_favorite = :is_favorite"
        params["is_favorite"] = is_favorite

    query += " ORDER BY id DESC"  # Новые книги сверху

    cursor.execute(query, params)
    books = cursor.fetchall()
    conn.close()

    return [dict(book) for book in books]


# --- READ (ONE) ---
@app.get("/books/{book_id}", summary="Получить книгу по ID")
def get_book(book_id: int):
    conn = get_db_connection()
    cursor = conn.cursor()

    cursor.execute("SELECT * FROM books WHERE id = :id", {"id": book_id})
    book = cursor.fetchone()
    conn.close()

    if not book:
        raise HTTPException(status_code=404, detail="Книга с таким ID не найдена")

    return dict(book)


# --- UPDATE ---
@app.put("/books/{book_id}", summary="Обновить данные книги")
def update_book(book_id: int, book_update: BookUpdate):
    conn = get_db_connection()
    cursor = conn.cursor()

    # 1. Проверяем, существует ли книга
    cursor.execute("SELECT id FROM books WHERE id = :id", {"id": book_id})
    if not cursor.fetchone():
        conn.close()
        raise HTTPException(status_code=404, detail="Книга с таким ID не найдена")

    # 2. Получаем только те поля, которые были переданы для обновления
    update_data = book_update.model_dump(exclude_unset=True)

    if not update_data:
        conn.close()
        raise HTTPException(status_code=400, detail="Нет данных для обновления")

    # 3. Формируем динамический SQL-запрос UPDATE
    set_clause = ", ".join([f"{key} = :{key}" for key in update_data.keys()])
    # Добавляем автоматическое обновление поля updated_at
    query = f"UPDATE books SET {set_clause}, updated_at = CURRENT_TIMESTAMP WHERE id = :id RETURNING *"

    update_data["id"] = book_id

    try:
        cursor.execute(query, update_data)
        updated_book = cursor.fetchone()
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=500, detail=f"Ошибка БД: {str(e)}")
    finally:
        conn.close()

    return dict(updated_book)


# --- DELETE ---
@app.delete("/books/{book_id}", summary="Удалить книгу")
def delete_book(book_id: int):
    conn = get_db_connection()
    cursor = conn.cursor()

    # Проверяем существование
    cursor.execute("SELECT id FROM books WHERE id = :id", {"id": book_id})
    if not cursor.fetchone():
        conn.close()
        raise HTTPException(status_code=404, detail="Книга с таким ID не найдена")

    # Удаляем
    cursor.execute("DELETE FROM books WHERE id = :id", {"id": book_id})
    conn.commit()
    conn.close()

    return {"message": "Книга успешно удалена", "id": book_id}