feat(sqlite): move from csv to sqlite

src/bookstore/inventory.py

@@ -1,4 +1,4 @@
-import csv
+import sqlite3
 from pathlib import Path
 
 from .book import Book
@@ -6,70 +6,65 @@ from .isbn import ISBN
 
 
 class Inventory:
-    def __init__(self, books: list[Book] | None) -> None:
+    def __init__(self, db_path: Path) -> None:
-        self.books: dict[ISBN, Book] = {book.isbn: book for book in books} if books else {}
+        self.conn = sqlite3.connect(db_path)
+        self.cursor = self.conn.cursor()
+        self.cursor.execute("CREATE TABLE IF NOT EXISTS Book (title TEXT, author TEXT, isbn TEXT PRIMARY KEY, price REAL, stock INTEGER)")
 
-    def read_from_file(self, path: Path) -> None:
+    def save(self) -> None:
-        """Read `Inventory` from `path` csv file."""
+        """Save `Inventory` to SQLite database."""
-        with open(path, mode="r") as file:
+        self.conn.commit()
-            reader = csv.reader(file)
 
-            next(reader, None)  # Skip header
+    def close(self) -> None:
+        """Close database connection."""
+        self.conn.close()
 
-            for title, author, isbn, price, stock in reader:
+    def add(self, *books: Book) -> None:
-                book = Book(title, author, isbn, price, stock)
-                self.add(book)
-
-    def save_to_file(self, path: Path) -> None:
-        """Save `Inventory` to `path` csv file."""
-        with open(path, mode="w") as file:
-            writer = csv.writer(file)
-
-            writer.writerow(["Title", "Auther", "ISBN", "Price", "Stock"])
-
-            for book in self.books.values():
-                writer.writerow([book.title, book.author, book.isbn, book.price, book.stock])
-
-    def add(self, book: Book) -> None:
         """Add `Book` to the `Inventory`. `Book`s ISBN must be unique."""
-        if book.isbn in self.books:
+        for book in books:
-            raise BaseException(f"Book with this ISBN: {book.isbn} already exists!")  # TODO: create custom exception
+            try:
-
+                self.cursor.execute("INSERT INTO Book VALUES (?, ?, ?, ?, ?)", (book.title, book.author, book.isbn, book.price, book.stock))
-        self.books[book.isbn] = book
+                self.conn.commit()
+            except sqlite3.InternalError:
+                print(f"A book with ISBN {book.isbn} already exists in the database.")
 
     def delete(self, isbn: ISBN) -> Book | None:
-        """Deletes `Book` from `Inventory` by `isbn` and returns deleted `Book`"""
+        """Deletes `Book` from `Inventory` by `ISBN` and returns deleted `Book`"""
-        return self.books.pop(isbn, None)
+        deleted_book = self.find_by_isbn(isbn)
+
+        self.cursor.execute("DELETE FROM Book WHERE isbn = ?", (isbn,))
+        self.conn.commit()
+
+        return deleted_book
 
     def find_by_isbn(self, isbn: ISBN) -> Book | None:
         """Looks up `Book` within `Inventory` by book `ISBN` and returns it. Returns `None` if it doesn't exist."""
-        if isbn not in self.books.keys():
+        self.cursor.execute("SELECT * FROM Book WHERE isbn = ?", (isbn,))
+        book = self.cursor.fetchone()
+        if not book:
             return None
-        return self.books.get(isbn)
+        return Book(*book)
 
     def find_by_title(self, title: str) -> list[Book] | None:
         """Looks up `Book`s within `Inventory` by book title and returns them as a `List`. Returns `None` if none were found"""
-        found_books = []
+        self.cursor.execute("SELECT * FROM Book WHERE title = ?", (title,))
-        for book in self.books.values():
+        books = self.cursor.fetchall()
-            if book.title.lower() == title.strip().lower():
+        if not books:
-                found_books.append(book)
-
-        if found_books:
-            return found_books
             return None
+        return [Book(*book) for book in books]
 
     def find_by_author(self, author: str) -> list[Book] | None:
         """Looks up `Book`s within `Inventory` by book author and returns them as a `List`. Returns `None` if none were found"""
-
+        self.cursor.execute("SELECT * FROM Book WHERE author = ?", (author,))
-        found_books = []
+        books = self.cursor.fetchall()
-        for book in self.books.values():
+        if not books:
-            if book.author.lower() == author.strip().lower():
-                found_books.append(book)
-
-        if found_books:
-            return found_books
             return None
+        return [Book(*book) for book in books]
 
-    def list_all(self) -> list[Book]:
+    def list_all(self) -> list[Book] | None:
         """Returns `List` of all `Book`s."""
-        return list(self.books.values())
+        self.cursor.execute("SELECT * FROM Book")
+        books = self.cursor.fetchall()
+        if not books:
+            return None
+        return [Book(*book) for book in books]