Files
CS374-Database-Systems/crud-operations/hw2.py
2025-10-02 10:08:38 -04:00

296 lines
9.0 KiB
Python

"""Practice writing CRUD queries using SQLite and Python's DB-API.
Author: Nicholas Tamassia
Version: 2025-09-08
"""
from pprint import pprint
import sqlite3
def connect(path):
"""Connect to the SQLite database file and return a cursor.
This function defines the global variables con and cur, which
you will use throughout the module.
THE CODE IS ALREADY FINISHED; DO NOT EDIT THIS FUNCTION.
Args:
path (str): File system path to the SQLite database file.
Returns:
sqlite3.Cursor: An object for executing SQL statements.
"""
global con, cur
con = sqlite3.connect(path)
cur = con.cursor()
cur.execute("PRAGMA foreign_keys = ON")
def create():
"""Execute SQL statements that create two tables of your choice.
Try to think of a unique example; no two students should have the
same table/column names and design. Your tables must have at least
five columns. At least one column must be an INTEGER, at least one
column must be a REAL number, and at least one column must be TEXT.
The first column of each table must be "id integer PRIMARY KEY".
Generally, this is not a good design, but it makes autograding the
assignment easier.
The second table must have a FOREIGN KEY reference to the first
table. Most but not all of the columns should be NOT NULL.
"""
cur.execute("""
CREATE TABLE IF NOT EXISTS Release (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
format TEXT NOT NULL,
sample_rate_khz REAL,
release_year INTEGER NOT NULL
)
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS Track (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
duration_sec REAL NOT NULL,
track_artist TEXT,
release_id INTEGER,
FOREIGN KEY (release_id) REFERENCES Release(id)
)
""")
def drop():
"""Execute SQL statements that drop your tables, if they exist.
This function is called at the beginning of the __main__ block so
that you can run the program over and over again to test your code.
"""
cur.execute("DROP TABLE IF EXISTS Track")
cur.execute("DROP TABLE IF EXISTS Release")
def insert_sample():
"""Execute SQL statements that insert three rows into each table.
The rows you insert should have realistic data values that show an
example of what the data might look like in each table. Please write
one INSERT statement that inserts three rows, not three statements
that insert one row each. At least one of the text values must have
an apostrophe (Ex: "JMU's mascot is the Duke Dog").
Note: The id values (in the first column) should be automatically
generated by SQLite. Do not include the id column in your code.
"""
cur.execute("""
INSERT INTO Release (title, format, sample_rate_khz, release_year)
VALUES
('In Praise of Shadows', '2xCD', 44.1, 2016),
('Queen A''rcadia', 'CD', 44.1, 2015),
('Gravity Noize', 'Digital Media', 44.1, 2018);
""")
cur.execute("""
INSERT INTO Track (title, duration_sec, track_artist, release_id)
VALUES
('Monochrome', 583.275, 'Proteus', 1),
('Bloody Duchess', 231.321, '葉月ゆら', 2),
('Storm Coming', 443.6743, 'Einhander', 1);
""")
def insert(title: str, format: float, sample_rate_khz: str, release_year: int):
"""Execute an SQL statement that inserts a row into the first table.
The values must be passed as parameters to this function. If your
table has 10 columns, then this function should have 9 parameters.
As a matter of style, the parameter and column names must match.
Note: The id value (in the first column) should be automatically
generated by SQLite. Do not include the id column in your code.
TODO Add parameters to the function definition and the docstring.
Args:
title (str): The title of the relase
format (str): The format of the relase
sample_rate_khs (float): The sample rate in kHz of the relase
release_year (int): The year the release was released
"""
cur.execute("""
INSERT INTO Release (title, format, sample_rate_khz, release_year)
VALUES (?, ?, ?, ?)
""", (title, format, sample_rate_khz, release_year))
def select_all() -> list[tuple]:
"""Execute an SQL statement that selects all rows from the first table.
Returns:
list[tuple]: Each tuple represents a row of the table.
"""
return cur.execute("""
SELECT * FROM Release
""").fetchall()
def select(pk: int) -> tuple:
"""Execute an SQL statement that selects one row from the first table.
Args:
pk (int): The id of the row to select.
Returns:
tuple: The row for the specified pk, or None if not found.
"""
return cur.execute("""
SELECT * FROM Release WHERE Release.id = ?
""", (pk,)).fetchone()
def update(id: int, title: str, format: str, sample_rate_khz: float, release_year: int):
"""Execute an SQL statement that updates one row in the first table.
The values must be passed as parameters to this function. If your
table has 10 columns, then this function should have 10 parameters.
As a matter of style, the parameter and column names must match.
TODO Add parameters to the function definition and the docstring.
Args:
id (int): The id of the release to update
title (str): The title of the relase
format (str): The format of the relase
sample_rate_khs (float): The sample rate in kHz of the relase
release_year (int): The year the release was released
"""
cur.execute("""
UPDATE Release
SET title = ?, format = ?, sample_rate_khz = ?, release_year = ?
WHERE id = ?
""", (title, format, sample_rate_khz, release_year, id))
def delete(pk):
"""Execute an SQL statement that deletes one row from the first table.
Note that, because of the foreign key, you cannot delete a row that is
referenced by the second table. If needed, modify the insert_sample()
function to insert at least one row that is not related to the second
table. Use that row's id to test this function.
Args:
pk (int): The id of the row to delete.
"""
cur.execute("""
DELETE FROM Release WHERE id = ?
""", (pk,))
def insert_invalid():
"""Cause the foreign key constraint to be violated by inserting a row.
The purpose of this function is to demonstrate your understanding of
foreign keys. Execute an SQL statement that inserts an invalid row into
the second table. The values should be hard-coded -- don't use question
marks in the query. SQLite should raise an IntegrityError when you call
the cur.execute() method.
"""
cur.execute("""
INSERT INTO Track (title, duration_sec, track_artist, release_id)
VALUES ('Personalizer', 473.2374, 'Electro.muster vo.みとせのりこ', 8);
""")
def update_invalid():
"""Cause the foreign key constraint to be violated by updating a row.
The purpose of this function is to demonstrate your understanding of
foreign keys. Execute an SQL statement that updates a row in the second
table. SQLite should raise an IntegrityError. The query values should be
hard-coded -- don't use question marks.
"""
cur.execute("""
UPDATE Track
SET release_id = 8
WHERE title = 'Storm Coming'
""")
def delete_invalid():
"""Cause the foreign key constraint to be violated by deleting a row.
The purpose of this function is to demonstrate your understanding of
foreign keys. Execute an SQL statement that deletes a row in the first
table. SQLite should raise an IntegrityError. The query values should be
hard-coded -- don't use question marks.
"""
cur.execute("""
DELETE FROM Release
WHERE title = 'In Praise of Shadows'
""")
if __name__ == "__main__":
# Feel free to modify the path to your SQLite database file.
# The autograder will use a different path for the database.
connect("hw2.db")
# The following code is provided to test your functions before submitting.
# TODO Be sure to add the arguments required for insert() and update().
drop()
create()
insert_sample()
pprint(select_all())
print()
insert("The Personalizer", "CD", 44.1, 2012)
pprint(select(2))
print()
update(4, "The Personalizer [リクエスト盤]", "CD", 44.1, 2012)
delete(4)
pprint(select_all())
print()
# The following tests make sure that your "invalid" functions work.
try:
insert_invalid()
print("insert_invalid doesn't work")
except sqlite3.IntegrityError:
pass
try:
update_invalid()
print("update_invalid doesn't work")
except sqlite3.IntegrityError:
pass
try:
delete_invalid()
print("delete_invalid doesn't work")
except sqlite3.IntegrityError:
pass