[main] WIP rework structure
This commit is contained in:
parent
6a42e522f9
commit
153da38f89
32
src/database/author.py
Normal file
32
src/database/author.py
Normal file
@ -0,0 +1,32 @@
|
|||||||
|
import logging
|
||||||
|
from typing import Dict
|
||||||
|
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from models import Author
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
def get_or_create_author(session: Session, author_data: Dict[str, str]) -> Author:
|
||||||
|
"""
|
||||||
|
Checks if an author exists in the database, creates one if not.
|
||||||
|
|
||||||
|
:param session: SQLAlchemy session object.
|
||||||
|
:param author_data: Dictionary containing author's first and last name.
|
||||||
|
:return: An Author instance (either existing or newly created).
|
||||||
|
"""
|
||||||
|
existing_author = session.query(Author).filter_by(
|
||||||
|
first_name=author_data["first_name"],
|
||||||
|
last_name=author_data["last_name"]
|
||||||
|
).one_or_none()
|
||||||
|
|
||||||
|
if existing_author is not None:
|
||||||
|
logger.debug(f"Author {author_data['first_name']} {author_data['last_name']} already exists. Reusing.")
|
||||||
|
return existing_author
|
||||||
|
|
||||||
|
logger.debug(f"Creating new author: {author_data['first_name']} {author_data['last_name']}")
|
||||||
|
author = Author(first_name=author_data["first_name"], last_name=author_data["last_name"])
|
||||||
|
session.add(author)
|
||||||
|
return author
|
||||||
|
|
||||||
|
__all__ = ["get_or_create_author"]
|
@ -1,16 +1,25 @@
|
|||||||
from typing import Dict, List
|
from typing import Dict, List, Optional
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, DatabaseError as SqlAlchemyDatabaseError
|
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, DatabaseError as SqlAlchemyDatabaseError
|
||||||
|
|
||||||
from utils.errors.database import DatabaseError, DuplicateEntryError, DatabaseConnectionError
|
from utils.errors.database import DatabaseError, DuplicateEntryError, DatabaseConnectionError
|
||||||
from models import Book
|
from models import Book
|
||||||
from database.manager import DatabaseManager
|
from database.manager import DatabaseManager
|
||||||
|
|
||||||
|
from .author import get_or_create_author
|
||||||
|
from .book_category import get_or_create_categories
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def fetch_all():
|
def fetch_all_books() -> List[Book]:
|
||||||
|
"""
|
||||||
|
Fetches all books from the database.
|
||||||
|
|
||||||
|
:return: A list of all books in the database.
|
||||||
|
:raises DatabaseConnectionError: If the connection to the database is interrupted.
|
||||||
|
:raises DatabaseError: If any other error occurs while fetching books.
|
||||||
|
"""
|
||||||
with DatabaseManager.get_session() as session:
|
with DatabaseManager.get_session() as session:
|
||||||
try:
|
try:
|
||||||
return session.query(Book).all()
|
return session.query(Book).all()
|
||||||
@ -19,47 +28,117 @@ def fetch_all():
|
|||||||
raise DatabaseConnectionError(
|
raise DatabaseConnectionError(
|
||||||
"Connection with database interrupted") from e
|
"Connection with database interrupted") from e
|
||||||
except SQLAlchemyError as e:
|
except SQLAlchemyError as e:
|
||||||
logger.error(f"An error occured when fetching all books: {e}")
|
logger.error(f"An error occurred when fetching all books: {e}")
|
||||||
raise DatabaseError(
|
raise DatabaseError(
|
||||||
"An error occured when fetching all books") from e
|
"An error occurred when fetching all books") from e
|
||||||
|
|
||||||
|
|
||||||
def create_new_book(book: Book):
|
def create_book(book: Dict[str, object]) -> None:
|
||||||
pass
|
"""
|
||||||
|
Creates a new book in the database.
|
||||||
|
|
||||||
|
:param book: A dictionary containing the book details (title, description, year_published, ISBN, author, and categories).
|
||||||
|
:raises DuplicateEntryError: If a book with the same ISBN already exists in the database.
|
||||||
|
:raises DatabaseConnectionError: If the connection to the database is interrupted.
|
||||||
|
:raises DatabaseError: If any other error occurs while creating the book.
|
||||||
|
"""
|
||||||
|
create_books([book])
|
||||||
|
|
||||||
|
|
||||||
def update_book(book: Book):
|
def create_books(books: List[Dict[str, object]]) -> None:
|
||||||
|
"""
|
||||||
|
Creates multiple books in the database.
|
||||||
|
|
||||||
|
:param books: A list of dictionaries, each containing the details of a book.
|
||||||
|
:raises DuplicateEntryError: If a book with the same ISBN already exists in the database.
|
||||||
|
:raises DatabaseConnectionError: If the connection to the database is interrupted.
|
||||||
|
:raises DatabaseError: If any other error occurs while creating the books.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
with DatabaseManager.get_session() as session:
|
with DatabaseManager.get_session() as session:
|
||||||
logger.debug(f"Updating book {book.title}")
|
for book in books:
|
||||||
existing_book = session.query(Book).get(book.id)
|
logger.debug(f"Attempting to create a new book: {book['title']}")
|
||||||
|
|
||||||
if not existing_book:
|
# Check if the book already exists
|
||||||
logger.warning(f"Book with id {book.id} not found")
|
existing_book = session.query(Book).filter_by(isbn=book["isbn"]).first()
|
||||||
raise DatabaseError("Book not found in the database")
|
|
||||||
|
|
||||||
existing_book.title = book.title
|
if existing_book:
|
||||||
existing_book.description = book.description
|
logger.warning(f"Book with ISBN {book['isbn']} already exists. Skipping.")
|
||||||
existing_book.year_published = book.year_published
|
raise DuplicateEntryError(f"Book with ISBN {book['isbn']} already exists.")
|
||||||
existing_book.isbn = book.isbn
|
|
||||||
|
author = get_or_create_author(session, book["author"])
|
||||||
|
categories = get_or_create_categories(session, book["categories"])
|
||||||
|
|
||||||
|
# Create the new book
|
||||||
|
new_book = Book(
|
||||||
|
title=book["title"],
|
||||||
|
description=book["description"],
|
||||||
|
year_published=book["year_published"],
|
||||||
|
isbn=book["isbn"],
|
||||||
|
author=author,
|
||||||
|
categories=categories
|
||||||
|
)
|
||||||
|
session.add(new_book)
|
||||||
|
|
||||||
session.commit()
|
session.commit()
|
||||||
logger.info(f"{book.title} successfully updated")
|
logger.info(f"Book {book['title']} successfully created.")
|
||||||
except IntegrityError as e:
|
except IntegrityError as e:
|
||||||
logger.warning("Data already exists")
|
logger.warning("Data already exists")
|
||||||
session.rollback()
|
raise DuplicateEntryError("Data already exists in the database") from e
|
||||||
raise DuplicateEntryError(
|
|
||||||
"Data already exists in the database") from e
|
|
||||||
except SqlAlchemyDatabaseError as e:
|
except SqlAlchemyDatabaseError as e:
|
||||||
session.rollback()
|
|
||||||
logger.critical("Connection with database interrupted")
|
logger.critical("Connection with database interrupted")
|
||||||
raise DatabaseConnectionError(
|
raise DatabaseConnectionError(
|
||||||
"Connection with database interrupted") from e
|
"Connection with database interrupted") from e
|
||||||
except SQLAlchemyError as e:
|
except SQLAlchemyError as e:
|
||||||
logger.error(f"An error occurred when saving book: {e}")
|
logger.error(f"An error occurred when creating the book: {e}")
|
||||||
session.rollback()
|
raise DatabaseError("An error occurred when creating the book") from e
|
||||||
raise DatabaseError(
|
|
||||||
"An error occurred when updating the book") from e
|
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["create_new_book", "update_book"]
|
def update_book(book: Dict[str, object]) -> None:
|
||||||
|
"""
|
||||||
|
Updates an existing book in the database. Reuses existing authors and categories if they exist.
|
||||||
|
|
||||||
|
:param book: A dictionary containing the updated book details, including the book ID.
|
||||||
|
:raises DatabaseError: If the book is not found in the database.
|
||||||
|
:raises DuplicateEntryError: If an attempt is made to update the book with duplicate data.
|
||||||
|
:raises DatabaseConnectionError: If the connection to the database is interrupted.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
with DatabaseManager.get_session() as session:
|
||||||
|
logger.debug(f"Updating book {book['title']}")
|
||||||
|
|
||||||
|
# Find the existing book
|
||||||
|
existing_book = session.query(Book).get(book["id"])
|
||||||
|
if not existing_book:
|
||||||
|
logger.warning(f"Book with ID {book['id']} not found")
|
||||||
|
raise DatabaseError("Book not found in the database")
|
||||||
|
|
||||||
|
# Get or create the author
|
||||||
|
author = get_or_create_author(session, book["author"])
|
||||||
|
|
||||||
|
# Get or create the categories
|
||||||
|
categories = get_or_create_categories(session, book["categories"])
|
||||||
|
|
||||||
|
# Update the book details
|
||||||
|
existing_book.title = book["title"]
|
||||||
|
existing_book.description = book["description"]
|
||||||
|
existing_book.year_published = book["year_published"]
|
||||||
|
existing_book.isbn = book["isbn"]
|
||||||
|
existing_book.author = author
|
||||||
|
existing_book.categories = categories
|
||||||
|
|
||||||
|
session.commit()
|
||||||
|
logger.info(f"{book['title']} successfully updated.")
|
||||||
|
except IntegrityError as e:
|
||||||
|
logger.warning("Data already exists")
|
||||||
|
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
|
||||||
|
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"]
|
||||||
|
39
src/database/book_category.py
Normal file
39
src/database/book_category.py
Normal file
@ -0,0 +1,39 @@
|
|||||||
|
from typing import Dict, List, Optional
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from models import BookCategory
|
||||||
|
from database.manager import DatabaseManager
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def get_or_create_categories(session, category_names: List[str]) -> List[BookCategory]:
|
||||||
|
"""
|
||||||
|
Checks if categories exist in the database, creates ones that don't.
|
||||||
|
|
||||||
|
:param session: SQLAlchemy session object.
|
||||||
|
:param category_names: List of category names.
|
||||||
|
:return: List of BookCategory instances (existing or newly created).
|
||||||
|
"""
|
||||||
|
processed_categories = {} # Cache for already processed categories
|
||||||
|
filtered_categories = []
|
||||||
|
|
||||||
|
for category_name in category_names:
|
||||||
|
if category_name in processed_categories:
|
||||||
|
filtered_categories.append(processed_categories[category_name])
|
||||||
|
continue
|
||||||
|
|
||||||
|
existing_category = session.query(BookCategory).filter_by(name=category_name).one_or_none()
|
||||||
|
|
||||||
|
if existing_category is not None:
|
||||||
|
logger.debug(f"Category {category_name} already exists. Reusing.")
|
||||||
|
processed_categories[category_name] = existing_category
|
||||||
|
filtered_categories.append(existing_category)
|
||||||
|
else:
|
||||||
|
logger.debug(f"Adding new category: {category_name}")
|
||||||
|
new_category = BookCategory(name=category_name)
|
||||||
|
session.add(new_category)
|
||||||
|
processed_categories[category_name] = new_category
|
||||||
|
filtered_categories.append(new_category)
|
||||||
|
|
||||||
|
return filtered_categories
|
24
src/database/book_overview.py
Normal file
24
src/database/book_overview.py
Normal file
@ -0,0 +1,24 @@
|
|||||||
|
from typing import Dict, List
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, DatabaseError as SqlAlchemyDatabaseError
|
||||||
|
|
||||||
|
from utils.errors.database import DatabaseError, DuplicateEntryError, DatabaseConnectionError
|
||||||
|
from models import BooksOverview
|
||||||
|
from database.manager import DatabaseManager
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_all() -> 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
|
||||||
|
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
|
@ -1,76 +0,0 @@
|
|||||||
from typing import Optional
|
|
||||||
import xml.etree.ElementTree as ET
|
|
||||||
from xml.dom import minidom
|
|
||||||
|
|
||||||
from database.manager import DatabaseManager
|
|
||||||
from utils.errors.export_error import ExportError
|
|
||||||
from utils.errors.no_export_entity_error import NoExportEntityError
|
|
||||||
|
|
||||||
from models import Book
|
|
||||||
|
|
||||||
|
|
||||||
class BookExporter():
|
|
||||||
def save_xml(self, file_path: str):
|
|
||||||
|
|
||||||
xml = self._get_full_xml()
|
|
||||||
|
|
||||||
with open(file_path, "w", encoding="utf-8") as file:
|
|
||||||
file.write(xml)
|
|
||||||
|
|
||||||
def _get_full_xml(self) -> str:
|
|
||||||
root = ET.Element("books")
|
|
||||||
|
|
||||||
with DatabaseManager.get_session() as session:
|
|
||||||
self.books = session.query(Book).all()
|
|
||||||
|
|
||||||
if not self.books:
|
|
||||||
raise NoExportEntityError("No books found to export")
|
|
||||||
|
|
||||||
for book in self.books:
|
|
||||||
# Create a <book> element
|
|
||||||
book_element = ET.SubElement(root, "book")
|
|
||||||
|
|
||||||
# Add <title>
|
|
||||||
title_element = ET.SubElement(book_element, "title")
|
|
||||||
title_element.text = book.title
|
|
||||||
|
|
||||||
# Add <author>
|
|
||||||
author_element = ET.SubElement(book_element, "author")
|
|
||||||
|
|
||||||
# Add <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.text = book.author.last_name
|
|
||||||
|
|
||||||
# Add <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.text = book.year_published
|
|
||||||
|
|
||||||
# Add <isbn>
|
|
||||||
isbn_element = ET.SubElement(book_element, "isbn")
|
|
||||||
isbn_element.text = book.isbn
|
|
||||||
|
|
||||||
# Add <categories>
|
|
||||||
categories_element = ET.SubElement(book_element, "categories")
|
|
||||||
for category in book.categories:
|
|
||||||
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))
|
|
||||||
return pretty_xml
|
|
@ -9,6 +9,7 @@ from xmlschema import XMLSchema
|
|||||||
from sqlalchemy.exc import IntegrityError
|
from sqlalchemy.exc import IntegrityError
|
||||||
|
|
||||||
from database.manager import DatabaseManager
|
from database.manager import DatabaseManager
|
||||||
|
from database.utils import get_or_create_categories, get_or_create_author
|
||||||
from utils.errors.import_error.xsd_scheme_not_found import XsdSchemeNotFoundError
|
from utils.errors.import_error.xsd_scheme_not_found import XsdSchemeNotFoundError
|
||||||
from utils.errors.import_error.invalid_contents_error import InvalidContentsError
|
from utils.errors.import_error.invalid_contents_error import InvalidContentsError
|
||||||
from utils.errors.import_error.import_error import ImportError
|
from utils.errors.import_error.import_error import ImportError
|
||||||
@ -68,77 +69,3 @@ class BookImporter:
|
|||||||
return books
|
return books
|
||||||
except ET.ParseError as e:
|
except ET.ParseError as e:
|
||||||
raise ImportError(f"Failed to parse XML file: {e}")
|
raise ImportError(f"Failed to parse XML file: {e}")
|
||||||
|
|
||||||
def save_books(self, books: List[Dict[str, object]]):
|
|
||||||
"""Saves a list of books to the database."""
|
|
||||||
try:
|
|
||||||
with DatabaseManager.get_session() as session:
|
|
||||||
processed_categories = {} # Cache for processed categories by name
|
|
||||||
|
|
||||||
for book_dict in books:
|
|
||||||
self.logger.debug(f"Attempting to save {book_dict['title']}")
|
|
||||||
|
|
||||||
# Check if the book already exists
|
|
||||||
existing_book = session.query(Book).filter_by(isbn=book_dict["isbn"]).first()
|
|
||||||
if existing_book:
|
|
||||||
self.logger.warning(f"ISBN {book_dict['isbn']} already exists. Skipping.")
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Check or add the author
|
|
||||||
existing_author = session.query(Author).filter_by(
|
|
||||||
first_name=book_dict["author"]["first_name"],
|
|
||||||
last_name=book_dict["author"]["last_name"]
|
|
||||||
).one_or_none()
|
|
||||||
|
|
||||||
if existing_author is not None:
|
|
||||||
self.logger.debug(f"Author {existing_author.first_name} {existing_author.last_name} already exists. Reusing.")
|
|
||||||
author = existing_author
|
|
||||||
else:
|
|
||||||
self.logger.debug(f"Creating new author: {book_dict['author']['first_name']} {book_dict['author']['last_name']}")
|
|
||||||
author = Author(
|
|
||||||
first_name=book_dict["author"]["first_name"],
|
|
||||||
last_name=book_dict["author"]["last_name"]
|
|
||||||
)
|
|
||||||
session.add(author)
|
|
||||||
|
|
||||||
# Handle categories
|
|
||||||
filtered_categories = []
|
|
||||||
for category_name in book_dict["categories"]:
|
|
||||||
if category_name in processed_categories:
|
|
||||||
filtered_categories.append(processed_categories[category_name])
|
|
||||||
continue
|
|
||||||
|
|
||||||
existing_category = session.query(BookCategory).filter_by(name=category_name).one_or_none()
|
|
||||||
if existing_category is not None:
|
|
||||||
self.logger.debug(f"Category {category_name} already exists. Reusing.")
|
|
||||||
processed_categories[category_name] = existing_category
|
|
||||||
filtered_categories.append(existing_category)
|
|
||||||
else:
|
|
||||||
self.logger.debug(f"Adding new category: {category_name}")
|
|
||||||
new_category = BookCategory(name=category_name)
|
|
||||||
session.add(new_category)
|
|
||||||
processed_categories[category_name] = new_category
|
|
||||||
filtered_categories.append(new_category)
|
|
||||||
|
|
||||||
book = Book(
|
|
||||||
title=book_dict["title"],
|
|
||||||
description=book_dict["description"],
|
|
||||||
year_published=book_dict["year_published"],
|
|
||||||
isbn=book_dict["isbn"],
|
|
||||||
author=author,
|
|
||||||
categories=filtered_categories
|
|
||||||
)
|
|
||||||
session.add(book)
|
|
||||||
user_config = UserConfig()
|
|
||||||
|
|
||||||
if user_config.simulate_slowdown:
|
|
||||||
self.logger.info("Simulating slowdown for 10 seconds")
|
|
||||||
time.sleep(10)
|
|
||||||
|
|
||||||
# Commit all changes
|
|
||||||
session.commit()
|
|
||||||
except e:
|
|
||||||
session.rollback()
|
|
||||||
raise ImportError(f"An error occurred when importing books: {e}") from e
|
|
||||||
finally:
|
|
||||||
session.close()
|
|
||||||
|
@ -16,5 +16,8 @@ class Author(Base):
|
|||||||
# Reference 'Book' as a string to avoid direct import
|
# Reference 'Book' as a string to avoid direct import
|
||||||
books = relationship('Book', back_populates='author')
|
books = relationship('Book', back_populates='author')
|
||||||
|
|
||||||
|
def to_dict(self):
|
||||||
|
return {col.name: getattr(self, col.name) for col in self.__table__.columns}
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["Author"]
|
__all__ = ["Author"]
|
||||||
|
@ -28,7 +28,17 @@ class Book(Base):
|
|||||||
|
|
||||||
author = relationship('Author', back_populates='books')
|
author = relationship('Author', back_populates='books')
|
||||||
categories = relationship('BookCategory',secondary='book_category_link',back_populates='books')
|
categories = relationship('BookCategory',secondary='book_category_link',back_populates='books')
|
||||||
book_category_statistics = relationship('BookCategoryStatistics', back_populates='book_category_statistics')
|
|
||||||
|
|
||||||
|
def to_dict(self):
|
||||||
|
book_dict = {col.name: getattr(self, col.name) for col in self.__table__.columns}
|
||||||
|
|
||||||
|
book_dict['author'] = {
|
||||||
|
'first_name': self.author.first_name,
|
||||||
|
'last_name': self.author.last_name
|
||||||
|
}
|
||||||
|
|
||||||
|
book_dict['categories'] = [category.name for category in self.categories]
|
||||||
|
|
||||||
|
return book_dict
|
||||||
|
|
||||||
__all__ = ["Book", "BookStatusEnum"]
|
__all__ = ["Book", "BookStatusEnum"]
|
||||||
|
@ -17,9 +17,11 @@ class BookCategory(Base):
|
|||||||
|
|
||||||
books = relationship(
|
books = relationship(
|
||||||
'Book',
|
'Book',
|
||||||
secondary='book_category_link', # Junction table
|
secondary='book_category_link',
|
||||||
back_populates='categories' # For bidirectional relationship
|
back_populates='categories',
|
||||||
)
|
)
|
||||||
|
book_category_statistics = relationship('BookCategoryStatistics', backref='book_category_statistics')
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["BookCategory"]
|
__all__ = ["BookCategory"]
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
from sqlalchemy import Column, Integer, ForeignKey
|
from sqlalchemy import Column, Integer, ForeignKey
|
||||||
|
|
||||||
|
from sqlalchemy.dialects.mysql import INTEGER
|
||||||
from sqlalchemy.orm import relationship
|
from sqlalchemy.orm import relationship
|
||||||
|
|
||||||
from .base import Base
|
from .base import Base
|
||||||
@ -7,11 +9,12 @@ class BookCategoryStatistics(Base):
|
|||||||
__tablename__ = 'book_category_statistics'
|
__tablename__ = 'book_category_statistics'
|
||||||
|
|
||||||
book_category_id = Column(Integer, ForeignKey('book_category.id', ondelete="cascade"), primary_key=True)
|
book_category_id = Column(Integer, ForeignKey('book_category.id', ondelete="cascade"), primary_key=True)
|
||||||
count = Column(Integer(unsigned=True), nullable=False, default=0)
|
count = Column(INTEGER(unsigned=True), nullable=False, default=0)
|
||||||
|
|
||||||
category = relationship(
|
category = relationship(
|
||||||
'BookCategory',
|
'BookCategory',
|
||||||
back_populates='book_category_statistics'
|
back_populates='book_category_statistics',
|
||||||
|
overlaps="book_category_statistics"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@ -24,5 +24,8 @@ class Member(Base):
|
|||||||
status = Column(Enum(MemberStatusEnum), nullable=True, default=MemberStatusEnum.active)
|
status = Column(Enum(MemberStatusEnum), nullable=True, default=MemberStatusEnum.active)
|
||||||
last_updated = Column(TIMESTAMP, nullable=True)
|
last_updated = Column(TIMESTAMP, nullable=True)
|
||||||
|
|
||||||
|
def to_dict(self):
|
||||||
|
return {col.name: getattr(self, col.name) for col in self.__table__.columns}
|
||||||
|
|
||||||
|
|
||||||
__all__ = ["Member", "MemberStatusEnum"]
|
__all__ = ["Member", "MemberStatusEnum"]
|
||||||
|
77
src/services/book_service.py
Normal file
77
src/services/book_service.py
Normal file
@ -0,0 +1,77 @@
|
|||||||
|
from typing import Optional, List
|
||||||
|
import xml.etree.ElementTree as ET
|
||||||
|
from xml.dom import minidom
|
||||||
|
|
||||||
|
from utils.errors import NoExportEntityError, ExportError, ExportFileError
|
||||||
|
|
||||||
|
from database import fetch_all_books
|
||||||
|
|
||||||
|
def export_to_xml(file_path: str):
|
||||||
|
all_books = fetch_all_books()
|
||||||
|
|
||||||
|
if not self.books:
|
||||||
|
raise NoExportEntityError("No books found to export")
|
||||||
|
|
||||||
|
xml = books_to_xml(all_books)
|
||||||
|
|
||||||
|
try:
|
||||||
|
with open(file_path, "w", encoding="utf-8") as file:
|
||||||
|
file.write(xml)
|
||||||
|
except OSError as e:
|
||||||
|
raise ExportFileError("Failed to save to a file") from e
|
||||||
|
|
||||||
|
|
||||||
|
def books_to_xml(books: List[Book]) -> str:
|
||||||
|
root = ET.Element("books")
|
||||||
|
|
||||||
|
for book in self.books:
|
||||||
|
# Create a <book> element
|
||||||
|
book_element = ET.SubElement(root, "book")
|
||||||
|
|
||||||
|
# Add <title>
|
||||||
|
title_element = ET.SubElement(book_element, "title")
|
||||||
|
title_element.text = book.title
|
||||||
|
|
||||||
|
# Add <author>
|
||||||
|
author_element = ET.SubElement(book_element, "author")
|
||||||
|
|
||||||
|
# Add <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.text = book.author.last_name
|
||||||
|
|
||||||
|
# Add <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.text = book.year_published
|
||||||
|
|
||||||
|
# Add <isbn>
|
||||||
|
isbn_element = ET.SubElement(book_element, "isbn")
|
||||||
|
isbn_element.text = book.isbn
|
||||||
|
|
||||||
|
# Add <categories>
|
||||||
|
categories_element = ET.SubElement(book_element, "categories")
|
||||||
|
for category in book.categories:
|
||||||
|
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))
|
||||||
|
return pretty_xml
|
||||||
|
|
||||||
|
__all__ = ["export_to_xml"]
|
@ -5,7 +5,7 @@ from PySide6.QtWidgets import (
|
|||||||
)
|
)
|
||||||
from PySide6.QtGui import QRegularExpressionValidator
|
from PySide6.QtGui import QRegularExpressionValidator
|
||||||
from PySide6.QtCore import QRegularExpression
|
from PySide6.QtCore import QRegularExpression
|
||||||
from models import Book, BookStatusEnum
|
from models import Book, BookStatusEnum, BookCategory
|
||||||
|
|
||||||
from database import update_book
|
from database import update_book
|
||||||
|
|
||||||
@ -25,6 +25,7 @@ class BookEditor(QDialog):
|
|||||||
self.create_new = False
|
self.create_new = False
|
||||||
self.fill_with_existing_data()
|
self.fill_with_existing_data()
|
||||||
else:
|
else:
|
||||||
|
self.book = Book()
|
||||||
self.logger.debug("Editing a new book")
|
self.logger.debug("Editing a new book")
|
||||||
self.create_new = True
|
self.create_new = True
|
||||||
|
|
||||||
@ -43,7 +44,7 @@ class BookEditor(QDialog):
|
|||||||
form_layout.addRow("Title:", self.title_input)
|
form_layout.addRow("Title:", self.title_input)
|
||||||
|
|
||||||
# Author field
|
# Author field
|
||||||
self.author_label = QLabel()
|
self.author_label = QLineEdit()
|
||||||
form_layout.addRow("Author: ", self.author_label)
|
form_layout.addRow("Author: ", self.author_label)
|
||||||
|
|
||||||
# Description field
|
# Description field
|
||||||
@ -57,13 +58,12 @@ class BookEditor(QDialog):
|
|||||||
|
|
||||||
# ISBN field
|
# ISBN field
|
||||||
self.isbn_input = QLineEdit()
|
self.isbn_input = QLineEdit()
|
||||||
self.isbn_expression = QRegularExpression("\d{10}|\d{13}")
|
self.isbn_expression = QRegularExpression("(\d{13})|(\d{10})")
|
||||||
self.isbn_validator = QRegularExpressionValidator(self.isbn_expression)
|
self.isbn_validator = QRegularExpressionValidator(self.isbn_expression)
|
||||||
self.isbn_input.setValidator(self.isbn_validator)
|
self.isbn_input.setValidator(self.isbn_validator)
|
||||||
form_layout.addRow("ISBN:", self.isbn_input)
|
form_layout.addRow("ISBN:", self.isbn_input)
|
||||||
|
|
||||||
self.categories_input = QLineEdit()
|
self.categories_input = QLineEdit()
|
||||||
self.categories_input.setEnabled(False)
|
|
||||||
form_layout.addRow("Categories: ", self.categories_input)
|
form_layout.addRow("Categories: ", self.categories_input)
|
||||||
|
|
||||||
layout.addLayout(form_layout)
|
layout.addLayout(form_layout)
|
||||||
@ -102,11 +102,21 @@ class BookEditor(QDialog):
|
|||||||
self.book.year_published = self.year_input.text()
|
self.book.year_published = self.year_input.text()
|
||||||
self.book.isbn = self.isbn_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:
|
try:
|
||||||
if self.create_new:
|
if self.create_new:
|
||||||
pass
|
pass
|
||||||
else:
|
else:
|
||||||
update_book(self.book)
|
update_book(self.book.to_dict())
|
||||||
|
|
||||||
|
QMessageBox.information(None,
|
||||||
|
"Success",
|
||||||
|
"Book updated successfully",
|
||||||
|
QMessageBox.StandardButton.Ok,
|
||||||
|
QMessageBox.StandardButton.NoButton)
|
||||||
|
|
||||||
self.accept()
|
self.accept()
|
||||||
except DuplicateEntryError as e:
|
except DuplicateEntryError as e:
|
||||||
QMessageBox.critical(None,
|
QMessageBox.critical(None,
|
||||||
|
2
src/ui/main_tabs/__init__.py
Normal file
2
src/ui/main_tabs/__init__.py
Normal file
@ -0,0 +1,2 @@
|
|||||||
|
from .book_overview_list import BookOverviewList
|
||||||
|
from .member_list import MemberList
|
7
src/ui/main_tabs/book_overview_list/__init__.py
Normal file
7
src/ui/main_tabs/book_overview_list/__init__.py
Normal file
@ -0,0 +1,7 @@
|
|||||||
|
from .overview_list import *
|
||||||
|
from .book_card import *
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
*overview_list.__all__,
|
||||||
|
*book_card.__all__
|
||||||
|
]
|
@ -154,3 +154,6 @@ class BookCard(QWidget):
|
|||||||
|
|
||||||
# Handle the response
|
# Handle the response
|
||||||
return response == QMessageBox.Yes
|
return response == QMessageBox.Yes
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["BookCard"]
|
@ -9,11 +9,12 @@ from .book_card import BookCard
|
|||||||
from models import BooksOverview
|
from models import BooksOverview
|
||||||
|
|
||||||
from database.manager import DatabaseManager
|
from database.manager import DatabaseManager
|
||||||
|
from database.book_overview import fetch_all
|
||||||
|
|
||||||
from ui.editor import MemberEditor
|
from ui.editor import MemberEditor
|
||||||
|
|
||||||
|
|
||||||
class LibraryDashboard(QWidget):
|
class BookOverviewList(QWidget):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
||||||
@ -98,7 +99,7 @@ class LibraryDashboard(QWidget):
|
|||||||
self.clear_layout(self.scroll_layout)
|
self.clear_layout(self.scroll_layout)
|
||||||
self.book_cards = []
|
self.book_cards = []
|
||||||
|
|
||||||
self.books = self.fetch_books_from_db()
|
self.books = fetch_all()
|
||||||
|
|
||||||
for book in self.books:
|
for book in self.books:
|
||||||
card = BookCard(book)
|
card = BookCard(book)
|
||||||
@ -106,13 +107,5 @@ class LibraryDashboard(QWidget):
|
|||||||
self.scroll_layout.addWidget(card)
|
self.scroll_layout.addWidget(card)
|
||||||
self.book_cards.append(card)
|
self.book_cards.append(card)
|
||||||
|
|
||||||
def fetch_books_from_db(self):
|
|
||||||
"""Fetch all books from the database."""
|
__all__ = ["BookOverviewList"]
|
||||||
try:
|
|
||||||
with DatabaseManager.get_session() as session:
|
|
||||||
books = session.query(BooksOverview).all()
|
|
||||||
return books
|
|
||||||
except Exception as e:
|
|
||||||
QMessageBox.critical(self, "Database Error",
|
|
||||||
f"Failed to fetch books: {e}")
|
|
||||||
return []
|
|
7
src/ui/main_tabs/member_list/__init__.py
Normal file
7
src/ui/main_tabs/member_list/__init__.py
Normal file
@ -0,0 +1,7 @@
|
|||||||
|
from .member_list import *
|
||||||
|
from .member_card import *
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
*member_list.__all__,
|
||||||
|
*member_card.__all__
|
||||||
|
]
|
@ -141,3 +141,6 @@ class MemberCard(QWidget):
|
|||||||
|
|
||||||
response = are_you_sure_box.exec()
|
response = are_you_sure_box.exec()
|
||||||
return response == QMessageBox.Yes
|
return response == QMessageBox.Yes
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["MemberCard"]
|
@ -112,3 +112,6 @@ class MemberList(QWidget):
|
|||||||
QMessageBox.critical(self, "Database Error",
|
QMessageBox.critical(self, "Database Error",
|
||||||
f"Failed to fetch members: {e}")
|
f"Failed to fetch members: {e}")
|
||||||
return []
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["MemberList"]
|
172
src/ui/menu_bar.py
Normal file
172
src/ui/menu_bar.py
Normal file
@ -0,0 +1,172 @@
|
|||||||
|
from PySide6.QtGui import QAction
|
||||||
|
from PySide6.QtWidgets import QMessageBox, QFileDialog, QMenuBar, QMenu, QDialog
|
||||||
|
from PySide6.QtCore import QStandardPaths
|
||||||
|
|
||||||
|
from ui.settings import SettingsDialog
|
||||||
|
|
||||||
|
from utils.errors import ExportError
|
||||||
|
|
||||||
|
class MenuBar(QMenuBar):
|
||||||
|
def __init__(self, parent):
|
||||||
|
super().__init__(parent)
|
||||||
|
|
||||||
|
self.parent = parent
|
||||||
|
self.file_types = {
|
||||||
|
"XML files (*.xml)": ".xml",
|
||||||
|
"Any file type (*)": ""
|
||||||
|
}
|
||||||
|
|
||||||
|
self.create_file_menu()
|
||||||
|
self.create_edit_menu()
|
||||||
|
self.create_help_menu()
|
||||||
|
|
||||||
|
def create_file_menu(self):
|
||||||
|
# File menu
|
||||||
|
file_menu = self.addMenu("File")
|
||||||
|
|
||||||
|
# New submenu
|
||||||
|
new_submenu = QMenu("New", self)
|
||||||
|
file_menu.addMenu(new_submenu)
|
||||||
|
|
||||||
|
# New book action
|
||||||
|
new_book_action = QAction("New book", self)
|
||||||
|
new_book_action.triggered.connect(self.new_book)
|
||||||
|
new_submenu.addAction(new_book_action)
|
||||||
|
|
||||||
|
# New member action
|
||||||
|
new_member_action = QAction("New member", self)
|
||||||
|
new_member_action.triggered.connect(self.new_member)
|
||||||
|
new_submenu.addAction(new_member_action)
|
||||||
|
|
||||||
|
# Import submenu
|
||||||
|
import_submenu = QMenu("Import", self)
|
||||||
|
file_menu.addMenu(import_submenu)
|
||||||
|
|
||||||
|
import_books_action = QAction("Import books", self)
|
||||||
|
import_books_action.triggered.connect(self.import_books)
|
||||||
|
import_submenu.addAction(import_books_action)
|
||||||
|
|
||||||
|
import_members_action = QAction("Import members", self)
|
||||||
|
import_members_action.triggered.connect(self.import_data)
|
||||||
|
import_submenu.addAction(import_members_action)
|
||||||
|
|
||||||
|
# Export submenu
|
||||||
|
export_submenu = QMenu("Export", self)
|
||||||
|
file_menu.addMenu(export_submenu)
|
||||||
|
|
||||||
|
export_overview_action = QAction("Export overview", self)
|
||||||
|
export_overview_action.triggered.connect(self.export_data)
|
||||||
|
export_submenu.addAction(export_overview_action)
|
||||||
|
|
||||||
|
export_books_action = QAction("Export books", self)
|
||||||
|
export_books_action.triggered.connect(self.export_books)
|
||||||
|
export_submenu.addAction(export_books_action)
|
||||||
|
|
||||||
|
export_members_action = QAction("Export members", self)
|
||||||
|
export_members_action.triggered.connect(self.export_data)
|
||||||
|
export_submenu.addAction(export_members_action)
|
||||||
|
|
||||||
|
file_menu.addSeparator()
|
||||||
|
|
||||||
|
exit_action = QAction("Exit", self)
|
||||||
|
exit_action.setShortcut("Ctrl+Q")
|
||||||
|
exit_action.triggered.connect(self.parent.close)
|
||||||
|
file_menu.addAction(exit_action)
|
||||||
|
|
||||||
|
def create_edit_menu(self):
|
||||||
|
# Edit menu
|
||||||
|
edit_menu = self.addMenu("Edit")
|
||||||
|
|
||||||
|
# Preferences menu
|
||||||
|
preferences_action = QAction("Preferences", self)
|
||||||
|
preferences_action.setShortcut("Ctrl+,")
|
||||||
|
preferences_action.triggered.connect(self.edit_preferences)
|
||||||
|
edit_menu.addAction(preferences_action)
|
||||||
|
|
||||||
|
def create_help_menu(self):
|
||||||
|
# Help menu
|
||||||
|
help_menu = self.addMenu("Help")
|
||||||
|
|
||||||
|
about_action = QAction("About", self)
|
||||||
|
about_action.triggered.connect(self.about)
|
||||||
|
help_menu.addAction(about_action)
|
||||||
|
|
||||||
|
def edit_preferences(self):
|
||||||
|
SettingsDialog(parent=self).exec()
|
||||||
|
|
||||||
|
def export_books(self):
|
||||||
|
try:
|
||||||
|
home_dir = QStandardPaths.writableLocation(
|
||||||
|
QStandardPaths.HomeLocation)
|
||||||
|
file_path, selected_filter = QFileDialog.getSaveFileName(self,
|
||||||
|
"Save book export",
|
||||||
|
home_dir,
|
||||||
|
";;".join(self.file_types.keys()))
|
||||||
|
if file_path:
|
||||||
|
selected_filetype = self.file_types[selected_filter]
|
||||||
|
|
||||||
|
if file_path.endswith(selected_filetype):
|
||||||
|
selected_filetype = ""
|
||||||
|
|
||||||
|
book_exporter = BookExporter()
|
||||||
|
book_exporter.save_xml(file_path + selected_filetype)
|
||||||
|
|
||||||
|
except OSError as e:
|
||||||
|
QMessageBox.critical(self,
|
||||||
|
"Error saving file",
|
||||||
|
f"Error occurred when saving the exported data: {e}",
|
||||||
|
QMessageBox.StandardButton.Ok,
|
||||||
|
QMessageBox.StandardButton.NoButton)
|
||||||
|
except ExportError as e:
|
||||||
|
QMessageBox.critical(self,
|
||||||
|
"Error exporting books",
|
||||||
|
f"An error occurred when exporting books: {e}",
|
||||||
|
QMessageBox.StandardButton.Ok,
|
||||||
|
QMessageBox.StandardButton.NoButton)
|
||||||
|
|
||||||
|
def import_books(self):
|
||||||
|
try:
|
||||||
|
home_dir = QStandardPaths.writableLocation(QStandardPaths.HomeLocation)
|
||||||
|
file_path, _ = QFileDialog.getOpenFileName(self, "Choose import file", home_dir, ";;".join(self.file_types.keys()))
|
||||||
|
|
||||||
|
if not file_path:
|
||||||
|
return # User canceled
|
||||||
|
|
||||||
|
importer = BookImporter()
|
||||||
|
books = importer.parse_xml(file_path)
|
||||||
|
|
||||||
|
if not books:
|
||||||
|
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)
|
||||||
|
QMessageBox.information(self, "Success", "Books imported successfully!", QMessageBox.Ok)
|
||||||
|
self.dashboard.redraw_cards()
|
||||||
|
else:
|
||||||
|
QMessageBox.information(self, "Canceled", "Import was canceled.", QMessageBox.Ok)
|
||||||
|
except ImportError as e:
|
||||||
|
QMessageBox.critical(self,
|
||||||
|
"Error importing books",
|
||||||
|
f"An error occurred when importing books from the file provided: {e}",
|
||||||
|
QMessageBox.StandardButton.Ok)
|
||||||
|
|
||||||
|
def new_book(self):
|
||||||
|
BookEditor().exec()
|
||||||
|
|
||||||
|
def new_member(self):
|
||||||
|
MemberEditor().exec()
|
||||||
|
|
||||||
|
def import_data(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def export_data(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
def about(self):
|
||||||
|
QMessageBox.information(
|
||||||
|
self, "About", "Library app demonstrating the phantom read problem")
|
@ -33,16 +33,20 @@ class SettingsDialog(QDialog):
|
|||||||
|
|
||||||
data_mode_layout.addWidget(self.data_mode_dropdown)
|
data_mode_layout.addWidget(self.data_mode_dropdown)
|
||||||
|
|
||||||
|
|
||||||
# Slowdown simulation
|
# Slowdown simulation
|
||||||
|
self.slowdown_layout = QtWidgets.QHBoxLayout()
|
||||||
|
|
||||||
self.slowdown_label = QtWidgets.QLabel(UserConfig.get_friendly_name("simulate_slowdown") + ":")
|
self.slowdown_label = QtWidgets.QLabel(UserConfig.get_friendly_name("simulate_slowdown") + ":")
|
||||||
data_mode_layout.addWidget(self.slowdown_label)
|
self.slowdown_layout.addWidget(self.slowdown_label)
|
||||||
|
|
||||||
self.slowdown_checkbox = QtWidgets.QCheckBox()
|
self.slowdown_checkbox = QtWidgets.QCheckBox()
|
||||||
self.slowdown_checkbox.setChecked(self.user_config.simulate_slowdown)
|
self.slowdown_checkbox.setChecked(self.user_config.simulate_slowdown)
|
||||||
|
|
||||||
data_mode_layout.addWidget(self.slowdown_checkbox)
|
self.slowdown_layout.addWidget(self.slowdown_checkbox)
|
||||||
|
|
||||||
layout.addLayout(data_mode_layout)
|
layout.addLayout(data_mode_layout)
|
||||||
|
layout.addLayout(self.slowdown_layout)
|
||||||
|
|
||||||
# Set the currently selected mode to the mode in UserConfig
|
# Set the currently selected mode to the mode in UserConfig
|
||||||
config = UserConfig()
|
config = UserConfig()
|
||||||
|
200
src/ui/window.py
200
src/ui/window.py
@ -1,25 +1,11 @@
|
|||||||
from PySide6.QtGui import QGuiApplication, QAction, QIcon
|
from PySide6.QtGui import QGuiApplication, QIcon
|
||||||
from PySide6.QtQml import QQmlApplicationEngine
|
from PySide6.QtWidgets import QMainWindow, QApplication, QTabWidget
|
||||||
from PySide6 import QtWidgets, QtCore
|
|
||||||
from PySide6.QtWidgets import QMessageBox, QFileDialog
|
|
||||||
from PySide6.QtCore import QStandardPaths
|
|
||||||
|
|
||||||
from ui.dashboard.dashboard import LibraryDashboard
|
from ui.main_tabs import BookOverviewList, MemberList
|
||||||
from ui.main_window_tabs.member_list.member_list import MemberList
|
from ui.menu_bar import MenuBar
|
||||||
from ui.editor import BookEditor, MemberEditor
|
|
||||||
|
|
||||||
from ui.settings import SettingsDialog
|
|
||||||
|
|
||||||
from ui.import_preview import PreviewDialog
|
|
||||||
|
|
||||||
from export.book_exporter import BookExporter
|
|
||||||
from importer.book.book_importer import BookImporter
|
|
||||||
|
|
||||||
from utils.errors.export_error import ExportError
|
|
||||||
from utils.errors.import_error.import_error import ImportError
|
|
||||||
|
|
||||||
|
|
||||||
class LibraryWindow(QtWidgets.QMainWindow):
|
class LibraryWindow(QMainWindow):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
super().__init__()
|
super().__init__()
|
||||||
|
|
||||||
@ -30,26 +16,20 @@ class LibraryWindow(QtWidgets.QMainWindow):
|
|||||||
|
|
||||||
self.center_window()
|
self.center_window()
|
||||||
|
|
||||||
# Set up menu bar
|
self.setMenuBar(MenuBar(self))
|
||||||
self.create_menu_bar()
|
|
||||||
|
|
||||||
# Central widget and layout
|
# Central widget and layout
|
||||||
central_widget = QtWidgets.QTabWidget()
|
central_widget = QTabWidget()
|
||||||
self.setCentralWidget(central_widget)
|
self.setCentralWidget(central_widget)
|
||||||
|
|
||||||
|
self.dashboard = BookOverviewList()
|
||||||
self.dashboard = LibraryDashboard()
|
|
||||||
self.member_list = MemberList()
|
self.member_list = MemberList()
|
||||||
central_widget.addTab(self.dashboard, "Dashboard")
|
central_widget.addTab(self.dashboard, "Dashboard")
|
||||||
central_widget.addTab(self.member_list, "Members")
|
central_widget.addTab(self.member_list, "Members")
|
||||||
|
|
||||||
self.file_types = {
|
|
||||||
"XML files (*.xml)": ".xml",
|
|
||||||
"Any file type (*)": ""}
|
|
||||||
|
|
||||||
def center_window(self):
|
def center_window(self):
|
||||||
# Get the screen geometry
|
# Get the screen geometry
|
||||||
screen = QtWidgets.QApplication.primaryScreen()
|
screen = QApplication.primaryScreen()
|
||||||
screen_geometry = screen.geometry()
|
screen_geometry = screen.geometry()
|
||||||
|
|
||||||
# Get the dimensions of the window
|
# Get the dimensions of the window
|
||||||
@ -63,165 +43,3 @@ class LibraryWindow(QtWidgets.QMainWindow):
|
|||||||
|
|
||||||
# Move the window to the calculated geometry
|
# Move the window to the calculated geometry
|
||||||
self.move(window_geometry.topLeft())
|
self.move(window_geometry.topLeft())
|
||||||
|
|
||||||
def create_menu_bar(self):
|
|
||||||
# Create the menu bar
|
|
||||||
menu_bar = self.menuBar()
|
|
||||||
|
|
||||||
# File menu
|
|
||||||
file_menu = menu_bar.addMenu("File")
|
|
||||||
|
|
||||||
# New submenu
|
|
||||||
new_submenu = QtWidgets.QMenu(self)
|
|
||||||
new_submenu.setTitle("New")
|
|
||||||
file_menu.addMenu(new_submenu)
|
|
||||||
|
|
||||||
# New book action
|
|
||||||
new_book_action = QAction("New book", self)
|
|
||||||
new_book_action.triggered.connect(self.new_book)
|
|
||||||
new_submenu.addAction(new_book_action)
|
|
||||||
|
|
||||||
# New book action
|
|
||||||
new_member_action = QAction("New member", self)
|
|
||||||
new_member_action.triggered.connect(self.new_member)
|
|
||||||
new_submenu.addAction(new_member_action)
|
|
||||||
|
|
||||||
# Import submenu
|
|
||||||
import_submenu = QtWidgets.QMenu(self)
|
|
||||||
import_submenu.setTitle("Import")
|
|
||||||
file_menu.addMenu(import_submenu)
|
|
||||||
|
|
||||||
import_books_action = QAction("Import books", self)
|
|
||||||
import_books_action.triggered.connect(self.import_books)
|
|
||||||
import_submenu.addAction(import_books_action)
|
|
||||||
|
|
||||||
import_members_action = QAction("Import members", self)
|
|
||||||
import_members_action.triggered.connect(self.import_data)
|
|
||||||
import_submenu.addAction(import_members_action)
|
|
||||||
|
|
||||||
# Export submenu
|
|
||||||
export_submenu = QtWidgets.QMenu(self)
|
|
||||||
export_submenu.setTitle("Export")
|
|
||||||
file_menu.addMenu(export_submenu)
|
|
||||||
|
|
||||||
# Export overview
|
|
||||||
export_overview_action = QAction("Export overview", self)
|
|
||||||
export_overview_action.triggered.connect(self.export_data)
|
|
||||||
export_submenu.addAction(export_overview_action)
|
|
||||||
|
|
||||||
# Export books
|
|
||||||
export_books_action = QAction("Export books", self)
|
|
||||||
export_books_action.triggered.connect(self.export_books)
|
|
||||||
export_submenu.addAction(export_books_action)
|
|
||||||
|
|
||||||
# Export members
|
|
||||||
export_members_action = QAction("Export members", self)
|
|
||||||
export_members_action.triggered.connect(self.export_data)
|
|
||||||
export_submenu.addAction(export_members_action)
|
|
||||||
|
|
||||||
file_menu.addSeparator()
|
|
||||||
|
|
||||||
exit_action = QAction("Exit", self)
|
|
||||||
exit_action.setShortcut("Ctrl+Q")
|
|
||||||
exit_action.triggered.connect(self.close)
|
|
||||||
file_menu.addAction(exit_action)
|
|
||||||
|
|
||||||
# Edit menu
|
|
||||||
edit_menu = menu_bar.addMenu("Edit")
|
|
||||||
|
|
||||||
# Preferences menu
|
|
||||||
preferences_action = QAction("Preferences", self)
|
|
||||||
preferences_action.setShortcut("Ctrl+,")
|
|
||||||
preferences_action.triggered.connect(self.edit_preferences)
|
|
||||||
edit_menu.addAction(preferences_action)
|
|
||||||
|
|
||||||
# Help menu
|
|
||||||
help_menu = menu_bar.addMenu("Help")
|
|
||||||
about_action = QAction("About", self)
|
|
||||||
about_action.triggered.connect(self.about)
|
|
||||||
help_menu.addAction(about_action)
|
|
||||||
|
|
||||||
# Menu action slots
|
|
||||||
def edit_preferences(self):
|
|
||||||
SettingsDialog(parent=self).exec()
|
|
||||||
|
|
||||||
# region Menu Actions
|
|
||||||
|
|
||||||
def export_books(self):
|
|
||||||
try:
|
|
||||||
home_dir = QStandardPaths.writableLocation(
|
|
||||||
QStandardPaths.HomeLocation)
|
|
||||||
file_path, selected_filter = QFileDialog.getSaveFileName(self,
|
|
||||||
"Save book export",
|
|
||||||
home_dir,
|
|
||||||
";;".join(self.file_types.keys()))
|
|
||||||
if file_path:
|
|
||||||
selected_filetype = self.file_types[selected_filter]
|
|
||||||
|
|
||||||
if file_path.endswith(selected_filetype):
|
|
||||||
selected_filetype = ""
|
|
||||||
|
|
||||||
book_exporter = BookExporter()
|
|
||||||
book_exporter.save_xml(file_path + selected_filetype)
|
|
||||||
|
|
||||||
except OSError as e:
|
|
||||||
QMessageBox.critical(self, "Error saving file", f"Error occurred when saving the exported data: {
|
|
||||||
e}", QMessageBox.StandardButton.Ok, QMessageBox.StandardButton.NoButton)
|
|
||||||
except ExportError as e:
|
|
||||||
QMessageBox.critical(self, "Error exporting books", f"An error occurred when exporting books: {
|
|
||||||
e}", QMessageBox.StandardButton.Ok, QMessageBox.StandardButton.NoButton)
|
|
||||||
|
|
||||||
def import_books(self):
|
|
||||||
try:
|
|
||||||
home_dir = QStandardPaths.writableLocation(
|
|
||||||
QStandardPaths.HomeLocation)
|
|
||||||
file_path, _ = QFileDialog.getOpenFileName(
|
|
||||||
self, "Choose import file", home_dir, ";;".join(
|
|
||||||
self.file_types.keys())
|
|
||||||
)
|
|
||||||
|
|
||||||
if not file_path:
|
|
||||||
return # User canceled
|
|
||||||
|
|
||||||
importer = BookImporter()
|
|
||||||
books = importer.parse_xml(file_path)
|
|
||||||
|
|
||||||
if not books:
|
|
||||||
QMessageBox.information(
|
|
||||||
self, "No New Books", "No new books to import.", QMessageBox.Ok)
|
|
||||||
return
|
|
||||||
|
|
||||||
# Show preview dialog
|
|
||||||
dialog = PreviewDialog(books, self)
|
|
||||||
if dialog.exec() == QtWidgets.QDialog.Accepted:
|
|
||||||
# User confirmed, proceed with importing
|
|
||||||
importer.save_books(books)
|
|
||||||
QMessageBox.information(
|
|
||||||
self, "Success", "Books imported successfully!", QMessageBox.Ok)
|
|
||||||
self.dashboard.redraw_cards()
|
|
||||||
else:
|
|
||||||
QMessageBox.information(
|
|
||||||
self, "Canceled", "Import was canceled.", QMessageBox.Ok)
|
|
||||||
except ImportError as e:
|
|
||||||
QMessageBox.critical(self, "Error importing books", f"An error occurred when importing books from the file provided: {
|
|
||||||
e}", QMessageBox.StandardButton.Ok, QMessageBox.StandardButton.NoButton)
|
|
||||||
|
|
||||||
def new_book(self):
|
|
||||||
# BookEditor()
|
|
||||||
pass
|
|
||||||
|
|
||||||
def new_member(self):
|
|
||||||
MemberEditor().exec()
|
|
||||||
|
|
||||||
|
|
||||||
def import_data(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
def export_data(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
def about(self):
|
|
||||||
QtWidgets.QMessageBox.information(
|
|
||||||
self, "About", "Library app demonstrating the phantom read problem")
|
|
||||||
|
|
||||||
# endregion
|
|
||||||
|
@ -0,0 +1,9 @@
|
|||||||
|
from .import_error import *
|
||||||
|
from .export_error import *
|
||||||
|
from .database import *
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
*import_error.__all__,
|
||||||
|
*export_error.__all__,
|
||||||
|
*database.__all__
|
||||||
|
]
|
@ -23,3 +23,6 @@ class DuplicateEntryError(DatabaseError):
|
|||||||
def __init__(self, message: str):
|
def __init__(self, message: str):
|
||||||
super().__init__(message)
|
super().__init__(message)
|
||||||
self.message = message
|
self.message = message
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["DatabaseError", "DatabaseConfigError", "DatabaseConnectionError", "DuplicateEntryError"]
|
||||||
|
@ -3,3 +3,19 @@ class ExportError(Exception):
|
|||||||
super().__init__(message)
|
super().__init__(message)
|
||||||
|
|
||||||
self.message = message
|
self.message = message
|
||||||
|
|
||||||
|
|
||||||
|
class NoExportEntityError(ExportError):
|
||||||
|
def __init__(self, message: str):
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
class ExportFileError(ExportError):
|
||||||
|
def __init__(self, message: str):
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["ExportError", "NoExportEntityError", "ExportFileError"]
|
22
src/utils/errors/import_error.py
Normal file
22
src/utils/errors/import_error.py
Normal file
@ -0,0 +1,22 @@
|
|||||||
|
class ImportError(Exception):
|
||||||
|
def __init__(self, message: str):
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
|
||||||
|
class InvalidContentsError(ImportError):
|
||||||
|
def __init__(self, message: str):
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
|
||||||
|
class XsdSchemeNotFoundError(ImportError):
|
||||||
|
def __init__(self, message: str):
|
||||||
|
super().__init__(message)
|
||||||
|
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
|
||||||
|
__all__ = ["ImportError", "InvalidContentsError", "XsdSchemeNotFoundError"]
|
@ -1,5 +0,0 @@
|
|||||||
class ImportError(Exception):
|
|
||||||
def __init__(self, message: str):
|
|
||||||
super().__init__(message)
|
|
||||||
|
|
||||||
self.message = message
|
|
@ -1,8 +0,0 @@
|
|||||||
from .import_error import ImportError
|
|
||||||
|
|
||||||
|
|
||||||
class InvalidContentsError(ImportError):
|
|
||||||
def __init__(self, message: str):
|
|
||||||
super().__init__(message)
|
|
||||||
|
|
||||||
self.message = message
|
|
@ -1,8 +0,0 @@
|
|||||||
from .import_error import ImportError
|
|
||||||
|
|
||||||
|
|
||||||
class XsdSchemeNotFoundError(ImportError):
|
|
||||||
def __init__(self, message: str):
|
|
||||||
super().__init__(message)
|
|
||||||
|
|
||||||
self.message = message
|
|
@ -1,6 +0,0 @@
|
|||||||
from .export_error import ExportError
|
|
||||||
class NoExportEntityError(ExportError):
|
|
||||||
def __init__(self, message: str):
|
|
||||||
super().__init__(message)
|
|
||||||
|
|
||||||
self.message = message
|
|
Loading…
x
Reference in New Issue
Block a user