books_sqlalchemy.py

import pandas
from sqlalchemy import *

try:
    engine = create_engine("mysql+pymysql://{user}:{pw}@localhost/"
                           .format(user="root",
                                   pw=""))
                                   
    DB_NAME = "booksdb"

    with engine.connect() as conn:
        conn.execute(f"DROP SCHEMA IF EXISTS {DB_NAME}")
        conn.execute(f"CREATE DATABASE {DB_NAME}")
        conn.execute(f"USE {DB_NAME}")
                                   
    metadata = MetaData()

    cls = Table('class', metadata,
        Column('class_id', Integer, primary_key=True, autoincrement=True),
        Column('class_name', String(24), unique=True, nullable=False)
    )

    author = Table('author', metadata,
        Column('author_id', Integer, primary_key=True, autoincrement=True),
        Column('author_fname', String(18), nullable=True),
        Column('author_lname', String(32), nullable=False),
        UniqueConstraint('author_fname', 'author_lname', name='uix_authname')
    )

    publisher = Table('publisher', metadata,
        Column('publisher_id', Integer, primary_key=True, autoincrement=True),
        Column('publisher_name', String(32), unique=True, nullable=False)
    )

    book = Table('book', metadata,
        Column('book_id', Integer, primary_key=True, autoincrement=True),
        Column('isbn10', CHAR(10), nullable=False),
        Column('isbn13', CHAR(13), nullable=False),
        Column('book_title', String(48), nullable=False),
        Column('book_currency', Enum("GBP", "USD", "EUR"), nullable=False),
        Column('book_rrp', DECIMAL(8,2), nullable=False),
        Column('book_price', DECIMAL(8,2), nullable=False),
        Column('book_stock', Integer, nullable=False),
        Column('book_format', Enum("Paperback", "Hardback", "Audio"), nullable=False),
        Column('book_pages', Integer, nullable=False),
        Column('book_publish_date', Date, nullable=False),
        Column('book_description', String(2048), nullable=False),
        Column('publisher_id_fk', Integer, ForeignKey("publisher.publisher_id"), nullable=False)
    )

    book_author = Table('book_author', metadata,
        Column('book_id_fk', Integer, ForeignKey("book.book_id"), primary_key=True),
        Column('author_id_fk', Integer, ForeignKey("author.author_id"), primary_key=True)
        
    )

    book_class = Table('book_class', metadata,
        Column('book_id_fk', Integer, ForeignKey("book.book_id"), primary_key=True),
        Column('class_id_fk', Integer, ForeignKey("class.class_id"), primary_key=True)
    )

    metadata.create_all(engine)
    
    conn = engine.connect()
    
    df = pandas.read_csv("publisher.csv")
    
    ins = publisher.insert()
    for index, row in df.iterrows():
        ins = publisher.insert().values(publisher_id=row['publisher_id'], publisher_name=row['publisher_name'])
        conn.execute(ins)
    
    df = pandas.read_csv("class.csv")
    
    ins = cls.insert()
    for index, row in df.iterrows():
        ins = cls.insert().values(class_id=row['class_id'], class_name=row['class_name'])
        conn.execute(ins)
    
    df = pandas.read_csv("author.csv")
    
    ins = author.insert()
    for index, row in df.iterrows():
        ins = author.insert().values(author_id=row['author_id'], author_fname=row['author_fname'], author_lname=row['author_lname'])
        conn.execute(ins)
    
    df = pandas.read_csv("book.csv")
    
    ins = book.insert()
    for index, row in df.iterrows():
        ins = book.insert().values(book_id=row['book_id'], isbn10=row['isbn-10'], isbn13=row['isbn-13'], book_title=row['book_title'], \
              book_currency=row['book_currency'], book_rrp=row['book_rrp'], book_price=row['book_price'], book_stock=row['book_stock'], \
              book_format=row['book_format'], book_pages=row['book_pages'], book_publish_date=row['book_publish_date'], \
              book_description=row['book_description'], publisher_id_fk=row['publisher_id_fk'])
        conn.execute(ins)
    
    df = pandas.read_csv("book_author.csv")
    
    ins = book_author.insert()
    for index, row in df.iterrows():
        ins = book_author.insert().values(book_id_fk=row['book_id_fk'], author_id_fk=row['author_id_fk'])
        conn.execute(ins)
    
    df = pandas.read_csv("book_class.csv")
    
    ins = book_class.insert()
    for index, row in df.iterrows():
        ins = book_class.insert().values(book_id_fk=row['book_id_fk'], class_id_fk=row['class_id_fk'])
        conn.execute(ins)

except Exception as err:
    print(f"Unexpected {err=}, {type(err)=}")
    exit(1)