db.py

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase
import csv
import os
from datetime import datetime as dt
from pathlib import Path
from sqlalchemy.exc import NoResultFound


class Base(DeclarativeBase):
    pass


db = SQLAlchemy(model_class=Base)


def initialize_database(Product, Category, Customer, Order, ProductOrder):
    try:

        db.session.execute(db.select(Product).limit(1)).all()
        db.session.execute(db.select(Category).limit(1)).all()
        db.session.execute(db.select(Customer).limit(1)).all()
        db.session.execute(db.select(Order).limit(1)).all()
        db.session.execute(db.select(ProductOrder).limit(1)).all()
        return False
    except:
        print("Creating database tables...")
        db.drop_all()
        db.create_all()
        
        base_path = Path(__file__).parent.absolute()
        csv_dir = os.path.join(base_path, 'csvs')
        

        categories = {}
        try:
            with open(os.path.join(csv_dir, 'products.csv'), 'r') as file:
                reader = csv.DictReader(file)
                # Extract unique category names
                for row in reader:
                    category_name = row['category']
                    if category_name and category_name not in categories:
                        category = Category(name=category_name)
                        db.session.add(category)
                        categories[category_name] = category
            
            db.session.flush()
            print("Categories extracted and added")
        except FileNotFoundError:
            print(f"Warning: products.csv not found at {os.path.join(csv_dir, 'products.csv')}")
        
        try:
            with open(os.path.join(csv_dir, 'customers.csv'), 'r') as file:
                reader = csv.DictReader(file)
                for row in reader:
                    customer = Customer(
                        nickname=row['nickname'],
                        email=row['email'],
                        created_at=dt.now()
                    )
                    db.session.add(customer)
            print("Customers added")
        except FileNotFoundError:
            print(f"Warning: customers.csv not found at {os.path.join(csv_dir, 'customers.csv')}")
        
        try:
            with open(os.path.join(csv_dir, 'products.csv'), 'r') as file:
                reader = csv.DictReader(file)
                for row in reader:
                    try:
                        category = categories.get(row['category'])
                        if category:
                            product = Product(
                                name=row['name'],
                                price=float(row['price']),
                                available=int(row['available']),
                                category_rel=category
                            )
                            db.session.add(product)
                    except Exception as e:
                        print(f"Error adding product {row.get('name', 'unknown')}: {e}")
            print("Products added")
        except FileNotFoundError:
            print(f"Warning: products.csv not found at {os.path.join(csv_dir, 'products.csv')}")
        
        db.session.commit()
        print("Database initialized successfully")
        return True
        
def import_data(Product, Category, Customer, Order, ProductOrder):
    if db.session.execute(db.select(db.func.count(Product.id))).scalar_one() > 0:
        return
    
    base_path = Path(__file__).parent.absolute()
    csv_dir = os.path.join(base_path, 'csvs')
    
    categories = {}
    try:
        with open(os.path.join(csv_dir, 'products.csv'), 'r') as file:
            reader = csv.DictReader(file)
            # Extract unique category names
            for row in reader:
                category_name = row['category']
                if category_name and category_name not in categories:
                    existing_category = db.session.execute(
                        db.select(Category).where(Category.name == category_name)
                    ).scalar_one_or_none()
                    
                    if not existing_category:
                        category = Category(name=category_name)
                        db.session.add(category)
                        db.session.flush()
                        categories[category_name] = category
                    else:
                        categories[category_name] = existing_category
        print("Categories processed")
    except FileNotFoundError:
        print(f"Warning: products.csv not found at {os.path.join(csv_dir, 'products.csv')}")
    
    try:
        with open(os.path.join(csv_dir, 'customers.csv'), 'r') as file:
            reader = csv.DictReader(file)
            for row in reader:
                existing_customer = db.session.execute(
                    db.select(Customer).where(Customer.email == row['email'])
                ).scalar_one_or_none()
                
                if not existing_customer:
                    customer = Customer(
                        nickname=row['nickname'],
                        email=row['email'],
                        created_at=dt.now()
                    )
                    db.session.add(customer)
        print("Customers processed")
    except FileNotFoundError:
        print(f"Warning: customers.csv not found at {os.path.join(csv_dir, 'customers.csv')}")
    
    try:
        with open(os.path.join(csv_dir, 'products.csv'), 'r') as file:
            reader = csv.DictReader(file)
            for row in reader:
                try:
                    existing_product = db.session.execute(
                        db.select(Product).where(Product.name == row['name'])
                    ).scalar_one_or_none()
                    
                    if not existing_product:
                        category = categories.get(row['category'])
                        if category:
                            product = Product(
                                name=row['name'],
                                price=float(row['price']),
                                available=int(row['available']),
                                category_rel=category
                            )
                            db.session.add(product)
                except Exception as e:
                    print(f"Error adding product {row.get('name', 'unknown')}: {e}")
        print("Products processed")
    except FileNotFoundError:
        print(f"Warning: products.csv not found at {os.path.join(csv_dir, 'products.csv')}")
    
    db.session.commit()
    print("Data import completed")
    
    return categories


def drop_all():
    print("Dropping all tables..")
    db.drop_all()
    print("All tables deleted successfully")


def create_all():
    print("Creating all tables...")
    db.create_all()
    print("All tables created successfully")