[main] Fixed imports and exports after structure rework
This commit is contained in:
parent
153da38f89
commit
d7ee79f7e9
5
src/assets/__init__.py
Normal file
5
src/assets/__init__.py
Normal file
@ -0,0 +1,5 @@
|
||||
from .asset_manager import *
|
||||
|
||||
__all__ = [
|
||||
*asset_manager.__all__
|
||||
]
|
13
src/assets/asset_manager.py
Normal file
13
src/assets/asset_manager.py
Normal file
@ -0,0 +1,13 @@
|
||||
import os
|
||||
|
||||
ASSETS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "assets")
|
||||
|
||||
|
||||
def get_asset(name: str) -> str:
|
||||
absolute_path = os.path.join(ASSETS_DIR, name)
|
||||
if not os.path.exists(absolute_path):
|
||||
raise FileNotFoundError(f"Asset not found: {absolute_path}")
|
||||
return os.path.abspath(absolute_path)
|
||||
|
||||
|
||||
__all__ = ["get_asset"]
|
@ -2,6 +2,7 @@ from typing import Dict, List, Optional
|
||||
import logging
|
||||
|
||||
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, DatabaseError as SqlAlchemyDatabaseError
|
||||
from sqlalchemy.orm import joinedload
|
||||
from utils.errors.database import DatabaseError, DuplicateEntryError, DatabaseConnectionError
|
||||
from models import Book
|
||||
from database.manager import DatabaseManager
|
||||
@ -22,15 +23,18 @@ def fetch_all_books() -> List[Book]:
|
||||
"""
|
||||
with DatabaseManager.get_session() as session:
|
||||
try:
|
||||
return session.query(Book).all()
|
||||
return session.query(Book) \
|
||||
.options(
|
||||
joinedload(Book.author),
|
||||
joinedload(Book.categories)
|
||||
) \
|
||||
.all()
|
||||
except SqlAlchemyDatabaseError as e:
|
||||
logger.critical("Connection with database interrupted")
|
||||
raise DatabaseConnectionError(
|
||||
"Connection with database interrupted") from e
|
||||
raise DatabaseConnectionError("Connection with database interrupted") from e
|
||||
except SQLAlchemyError as e:
|
||||
logger.error(f"An error occurred when fetching all books: {e}")
|
||||
raise DatabaseError(
|
||||
"An error occurred when fetching all books") from e
|
||||
raise DatabaseError("An error occurred when fetching all books") from e
|
||||
|
||||
|
||||
def create_book(book: Dict[str, object]) -> None:
|
||||
@ -64,7 +68,7 @@ def create_books(books: List[Dict[str, object]]) -> None:
|
||||
|
||||
if existing_book:
|
||||
logger.warning(f"Book with ISBN {book['isbn']} already exists. Skipping.")
|
||||
raise DuplicateEntryError(f"Book with ISBN {book['isbn']} already exists.")
|
||||
continue
|
||||
|
||||
author = get_or_create_author(session, book["author"])
|
||||
categories = get_or_create_categories(session, book["categories"])
|
||||
@ -87,8 +91,7 @@ def create_books(books: List[Dict[str, object]]) -> None:
|
||||
raise DuplicateEntryError("Data already exists in the database") from e
|
||||
except SqlAlchemyDatabaseError as e:
|
||||
logger.critical("Connection with database interrupted")
|
||||
raise DatabaseConnectionError(
|
||||
"Connection with database interrupted") from e
|
||||
raise DatabaseConnectionError("Connection with database interrupted") from e
|
||||
except SQLAlchemyError as e:
|
||||
logger.error(f"An error occurred when creating the book: {e}")
|
||||
raise DatabaseError("An error occurred when creating the book") from e
|
||||
@ -134,11 +137,10 @@ def update_book(book: Dict[str, object]) -> None:
|
||||
raise DuplicateEntryError("Data already exists in the database") from e
|
||||
except SqlAlchemyDatabaseError as e:
|
||||
logger.critical("Connection with database interrupted")
|
||||
raise DatabaseConnectionError(
|
||||
"Connection with database interrupted") from e
|
||||
raise DatabaseConnectionError("Connection with database interrupted") from e
|
||||
except SQLAlchemyError as e:
|
||||
logger.error(f"An error occurred when updating the book: {e}")
|
||||
raise DatabaseError("An error occurred when updating the book") from e
|
||||
|
||||
|
||||
__all__ = ["create_book", "create_books", "update_book", "fetch_all"]
|
||||
__all__ = ["create_book", "create_books", "update_book", "fetch_all_books"]
|
||||
|
@ -10,15 +10,16 @@ from database.manager import DatabaseManager
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def fetch_all() -> List[BooksOverview]:
|
||||
def fetch_all_book_overviews() -> List[BooksOverview]:
|
||||
with DatabaseManager.get_session() as session:
|
||||
try:
|
||||
return session.query(BooksOverview).all()
|
||||
except SqlAlchemyDatabaseError as e:
|
||||
logger.critical("Connection with database interrupted")
|
||||
raise DatabaseConnectionError(
|
||||
"Connection with database interrupted") from e
|
||||
raise DatabaseConnectionError("Connection with database interrupted") from e
|
||||
except SQLAlchemyError as e:
|
||||
logger.error(f"An error occured when fetching all books: {e}")
|
||||
raise DatabaseError(
|
||||
"An error occured when fetching all books") from e
|
||||
raise DatabaseError("An error occured when fetching all books") from e
|
||||
|
||||
|
||||
__all__ = ["fetch_all_book_overviews"]
|
||||
|
@ -1,15 +1,38 @@
|
||||
from typing import Optional, List
|
||||
import os
|
||||
import logging
|
||||
from typing import Optional, List, Dict
|
||||
import xml.etree.ElementTree as ET
|
||||
from xml.dom import minidom
|
||||
from xmlschema import XMLSchema
|
||||
|
||||
from utils.errors import NoExportEntityError, ExportError, ExportFileError
|
||||
from utils.errors import (
|
||||
NoExportEntityError,
|
||||
ExportError,
|
||||
ExportFileError,
|
||||
InvalidContentsError,
|
||||
XsdSchemeNotFoundError,
|
||||
ImportError,
|
||||
)
|
||||
from models import Book
|
||||
from database import fetch_all_books, create_books
|
||||
from assets import asset_manager
|
||||
|
||||
from database import fetch_all_books
|
||||
# Initialize logger and XML Schema
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def export_to_xml(file_path: str):
|
||||
|
||||
try:
|
||||
logger.debug("Loading XSD schema")
|
||||
SCHEMA = XMLSchema(asset_manager.get_asset("book_import_scheme.xsd"))
|
||||
except Exception as e:
|
||||
logger.error("Failed to load XSD schema")
|
||||
raise XsdSchemeNotFoundError(f"Failed to load XSD schema: {e}")
|
||||
|
||||
|
||||
def export_to_xml(file_path: str) -> None:
|
||||
all_books = fetch_all_books()
|
||||
|
||||
if not self.books:
|
||||
if not all_books:
|
||||
raise NoExportEntityError("No books found to export")
|
||||
|
||||
xml = books_to_xml(all_books)
|
||||
@ -20,11 +43,55 @@ def export_to_xml(file_path: str):
|
||||
except OSError as e:
|
||||
raise ExportFileError("Failed to save to a file") from e
|
||||
|
||||
def save_books(books: List[Dict[str, object]]):
|
||||
create_books(books)
|
||||
|
||||
def parse_books_from_xml(file_path: str) -> List[Dict[str, object]]:
|
||||
if not SCHEMA.is_valid(file_path):
|
||||
raise InvalidContentsError("XML file is not valid according to XSD schema.")
|
||||
|
||||
try:
|
||||
tree = ET.parse(file_path)
|
||||
root = tree.getroot()
|
||||
|
||||
books = []
|
||||
for book_element in root.findall("book"):
|
||||
title = book_element.find("title").text
|
||||
year_published = book_element.find("year_published").text
|
||||
description = book_element.find("description").text
|
||||
isbn = book_element.find("isbn").text
|
||||
|
||||
# Parse author
|
||||
author_element = book_element.find("author")
|
||||
author = {
|
||||
"first_name": author_element.find("first_name").text,
|
||||
"last_name": author_element.find("last_name").text,
|
||||
}
|
||||
|
||||
# Parse categories
|
||||
category_elements = book_element.find("categories").findall("category")
|
||||
categories = [category_element.text for category_element in category_elements]
|
||||
|
||||
# Create a book dictionary
|
||||
book = {
|
||||
"title" : title,
|
||||
"description" : description,
|
||||
"year_published" : year_published,
|
||||
"isbn" : isbn,
|
||||
"author" : author,
|
||||
"categories" : categories,
|
||||
}
|
||||
books.append(book)
|
||||
|
||||
return books
|
||||
except ET.ParseError as e:
|
||||
raise ImportError(f"Failed to parse XML file: {e}")
|
||||
|
||||
|
||||
def books_to_xml(books: List[Book]) -> str:
|
||||
root = ET.Element("books")
|
||||
|
||||
for book in self.books:
|
||||
for book in books:
|
||||
# Create a <book> element
|
||||
book_element = ET.SubElement(root, "book")
|
||||
|
||||
@ -36,22 +103,18 @@ def books_to_xml(books: List[Book]) -> str:
|
||||
author_element = ET.SubElement(book_element, "author")
|
||||
|
||||
# Add <first_name>
|
||||
author_first_name_element = ET.SubElement(
|
||||
author_element, "first_name")
|
||||
author_first_name_element = ET.SubElement(author_element, "first_name")
|
||||
author_first_name_element.text = book.author.first_name
|
||||
|
||||
author_last_name_element = ET.SubElement(
|
||||
author_element, "last_name")
|
||||
author_last_name_element = ET.SubElement(author_element, "last_name")
|
||||
author_last_name_element.text = book.author.last_name
|
||||
|
||||
# Add <description>
|
||||
description_element = ET.SubElement(
|
||||
book_element, "description")
|
||||
description_element = ET.SubElement(book_element, "description")
|
||||
description_element.text = book.description
|
||||
|
||||
# Add <year_published>
|
||||
year_published_element = ET.SubElement(
|
||||
book_element, "year_published")
|
||||
year_published_element = ET.SubElement(book_element, "year_published")
|
||||
year_published_element.text = book.year_published
|
||||
|
||||
# Add <isbn>
|
||||
@ -61,17 +124,15 @@ def books_to_xml(books: List[Book]) -> str:
|
||||
# Add <categories>
|
||||
categories_element = ET.SubElement(book_element, "categories")
|
||||
for category in book.categories:
|
||||
category_element = ET.SubElement(
|
||||
categories_element, "category")
|
||||
category_element = ET.SubElement(categories_element, "category")
|
||||
category_element.text = category.name
|
||||
|
||||
|
||||
# Convert the tree to a string
|
||||
tree_str = ET.tostring(root, encoding="unicode")
|
||||
|
||||
# Pretty print the XML
|
||||
pretty_xml = minidom.parseString(
|
||||
tree_str).toprettyxml(indent=(" " * 4))
|
||||
pretty_xml = minidom.parseString(tree_str).toprettyxml(indent=(" " * 4))
|
||||
return pretty_xml
|
||||
|
||||
__all__ = ["export_to_xml"]
|
||||
|
||||
__all__ = ["export_to_xml", "parse_books_from_xml"]
|
||||
|
@ -1,3 +1,5 @@
|
||||
from typing import Dict
|
||||
|
||||
import logging
|
||||
|
||||
from PySide6.QtWidgets import (
|
||||
@ -7,7 +9,7 @@ from PySide6.QtGui import QRegularExpressionValidator
|
||||
from PySide6.QtCore import QRegularExpression
|
||||
from models import Book, BookStatusEnum, BookCategory
|
||||
|
||||
from database import update_book
|
||||
from database import update_book, create_book
|
||||
|
||||
from utils.errors.database import DatabaseError, DatabaseConnectionError, DuplicateEntryError
|
||||
|
||||
@ -21,11 +23,9 @@ class BookEditor(QDialog):
|
||||
|
||||
if book:
|
||||
self.logger.debug(f"Editing existing book {book.title}")
|
||||
self.book = book
|
||||
self.create_new = False
|
||||
self.fill_with_existing_data()
|
||||
self.fill_with_existing_data(book)
|
||||
else:
|
||||
self.book = Book()
|
||||
self.logger.debug("Editing a new book")
|
||||
self.create_new = True
|
||||
|
||||
@ -81,35 +81,26 @@ class BookEditor(QDialog):
|
||||
|
||||
layout.addLayout(button_layout)
|
||||
|
||||
def fill_with_existing_data(self):
|
||||
self.title_input.setText(self.book.title)
|
||||
self.description_input.setText(self.book.description)
|
||||
self.year_input.setText(self.book.year_published)
|
||||
self.isbn_input.setText(self.book.isbn)
|
||||
def fill_with_existing_data(self, book: Book):
|
||||
self.title_input.setText(book.title)
|
||||
self.description_input.setText(book.description)
|
||||
self.year_input.setText(book.year_published)
|
||||
self.isbn_input.setText(book.isbn)
|
||||
|
||||
full_author_name = f"{self.book.author.first_name} {
|
||||
self.book.author.last_name}"
|
||||
full_author_name = f"{book.author.first_name} {book.author.last_name}"
|
||||
self.author_label.setText(full_author_name)
|
||||
|
||||
all_categories = ", ".join(
|
||||
category.name for category in self.book.categories)
|
||||
all_categories = ", ".join(category.name for category in book.categories)
|
||||
self.categories_input.setText(all_categories)
|
||||
|
||||
def save_book(self):
|
||||
# Update book object with input values
|
||||
self.book.title = self.title_input.text()
|
||||
self.book.description = self.description_input.toPlainText()
|
||||
self.book.year_published = self.year_input.text()
|
||||
self.book.isbn = self.isbn_input.text()
|
||||
|
||||
categories_list = self.categories_input.text().split(",")
|
||||
self.book.categories = [BookCategory(name=category.strip()) for category in categories_list]
|
||||
|
||||
try:
|
||||
book_object = self.parse_inputs()
|
||||
|
||||
if self.create_new:
|
||||
pass
|
||||
create_book(book_object)
|
||||
else:
|
||||
update_book(self.book.to_dict())
|
||||
update_book(book_object)
|
||||
|
||||
QMessageBox.information(None,
|
||||
"Success",
|
||||
@ -118,6 +109,12 @@ class BookEditor(QDialog):
|
||||
QMessageBox.StandardButton.NoButton)
|
||||
|
||||
self.accept()
|
||||
except ValueError as e:
|
||||
QMessageBox.critical(None,
|
||||
"Invalid Input",
|
||||
f"Input validation failed: {e}",
|
||||
QMessageBox.StandardButton.Ok,
|
||||
QMessageBox.StandardButton.NoButton)
|
||||
except DuplicateEntryError as e:
|
||||
QMessageBox.critical(None,
|
||||
"ISBN is already in use",
|
||||
@ -132,10 +129,59 @@ class BookEditor(QDialog):
|
||||
QMessageBox.StandardButton.NoButton)
|
||||
except DatabaseError as e:
|
||||
QMessageBox.critical(self.parent,
|
||||
"An error occured",
|
||||
"An error occurred",
|
||||
f"Could not save the book because of the following error: {e}",
|
||||
QMessageBox.StandardButton.Ok,
|
||||
QMessageBox.StandardButton.NoButton)
|
||||
|
||||
def parse_inputs(self) -> Dict[str, object]:
|
||||
# Title validation
|
||||
title = self.title_input.text().strip()
|
||||
if not title or len(title) > 100:
|
||||
raise ValueError("Title must be non-empty and at most 100 characters long.")
|
||||
|
||||
# Author validation
|
||||
author_name = self.author_label.text().strip()
|
||||
if not author_name or len(author_name.split()) < 2:
|
||||
raise ValueError("Author must include at least a first and last name.")
|
||||
|
||||
# Split author name into first and last names
|
||||
author_parts = author_name.split()
|
||||
first_name = author_parts[0]
|
||||
last_name = " ".join(author_parts[1:])
|
||||
|
||||
# Description validation
|
||||
description = self.description_input.toPlainText().strip()
|
||||
if not description:
|
||||
raise ValueError("Description cannot be empty.")
|
||||
|
||||
# Year published validation
|
||||
year_published = self.year_input.text().strip()
|
||||
if not year_published.isdigit() or len(year_published) != 4 or int(year_published) < 0:
|
||||
raise ValueError("Year published must be a 4-digit positive number.")
|
||||
|
||||
# ISBN validation
|
||||
isbn = self.isbn_input.text().strip()
|
||||
if not isbn or len(isbn) not in (10, 13):
|
||||
raise ValueError("ISBN must be either 10 or 13 characters long.")
|
||||
|
||||
# Categories validation
|
||||
category_text = self.categories_input.text().strip()
|
||||
categories = [category.strip() for category in category_text.split(",") if category.strip()]
|
||||
if not categories:
|
||||
raise ValueError("At least one category must be specified.")
|
||||
|
||||
# Map parsed values to dictionary format for saving
|
||||
return {
|
||||
"title": title,
|
||||
"author": {
|
||||
"first_name": first_name,
|
||||
"last_name": last_name
|
||||
},
|
||||
"description": description,
|
||||
"year_published": year_published,
|
||||
"isbn": isbn,
|
||||
"categories": categories
|
||||
}
|
||||
|
||||
__all__ = ["BookEditor"]
|
||||
|
@ -9,7 +9,7 @@ from .book_card import BookCard
|
||||
from models import BooksOverview
|
||||
|
||||
from database.manager import DatabaseManager
|
||||
from database.book_overview import fetch_all
|
||||
from database.book_overview import fetch_all_book_overviews
|
||||
|
||||
from ui.editor import MemberEditor
|
||||
|
||||
@ -99,7 +99,7 @@ class BookOverviewList(QWidget):
|
||||
self.clear_layout(self.scroll_layout)
|
||||
self.book_cards = []
|
||||
|
||||
self.books = fetch_all()
|
||||
self.books = fetch_all_book_overviews()
|
||||
|
||||
for book in self.books:
|
||||
card = BookCard(book)
|
||||
|
@ -3,8 +3,11 @@ from PySide6.QtWidgets import QMessageBox, QFileDialog, QMenuBar, QMenu, QDialog
|
||||
from PySide6.QtCore import QStandardPaths
|
||||
|
||||
from ui.settings import SettingsDialog
|
||||
from ui.import_preview import PreviewDialog
|
||||
from ui.editor import BookEditor, MemberEditor
|
||||
|
||||
from utils.errors import ExportError
|
||||
from utils.errors import ExportError, ExportFileError
|
||||
from services import book_service
|
||||
|
||||
class MenuBar(QMenuBar):
|
||||
def __init__(self, parent):
|
||||
@ -107,11 +110,10 @@ class MenuBar(QMenuBar):
|
||||
|
||||
if file_path.endswith(selected_filetype):
|
||||
selected_filetype = ""
|
||||
|
||||
book_service.export_to_xml(file_path + selected_filetype)
|
||||
|
||||
book_exporter = BookExporter()
|
||||
book_exporter.save_xml(file_path + selected_filetype)
|
||||
|
||||
except OSError as e:
|
||||
except ExportFileError as e:
|
||||
QMessageBox.critical(self,
|
||||
"Error saving file",
|
||||
f"Error occurred when saving the exported data: {e}",
|
||||
@ -132,21 +134,18 @@ class MenuBar(QMenuBar):
|
||||
if not file_path:
|
||||
return # User canceled
|
||||
|
||||
importer = BookImporter()
|
||||
books = importer.parse_xml(file_path)
|
||||
|
||||
books = book_service.parse_books_from_xml(file_path)
|
||||
if not books:
|
||||
QMessageBox.information(
|
||||
self, "No New Books", "No new books to import.", QMessageBox.Ok)
|
||||
QMessageBox.information(self, "No New Books", "No new books to import.", QMessageBox.Ok)
|
||||
return
|
||||
|
||||
# Show preview dialog
|
||||
dialog = PreviewDialog(books, self)
|
||||
if dialog.exec() == QDialog.Accepted:
|
||||
# User confirmed, proceed with importing
|
||||
create_books(books)
|
||||
book_service.create_books(books)
|
||||
QMessageBox.information(self, "Success", "Books imported successfully!", QMessageBox.Ok)
|
||||
self.dashboard.redraw_cards()
|
||||
self.parent.redraw_book_cards()
|
||||
else:
|
||||
QMessageBox.information(self, "Canceled", "Import was canceled.", QMessageBox.Ok)
|
||||
except ImportError as e:
|
||||
|
@ -43,3 +43,6 @@ class LibraryWindow(QMainWindow):
|
||||
|
||||
# Move the window to the calculated geometry
|
||||
self.move(window_geometry.topLeft())
|
||||
|
||||
def redraw_book_cards(self):
|
||||
self.dashboard.redraw_cards()
|
Loading…
x
Reference in New Issue
Block a user