Files
CS374-Database-Systems/Python-DB-API/vdoe.py
2025-10-07 21:05:38 -04:00

142 lines
3.9 KiB
Python

"""Analyze SOL test pass rates and other results provided by VDOE.
Author: Nicholas Tamassia
Version: 2.9.2025
"""
import csv
import sqlite3
# TODO add your functions here (see the lab instructions)
def create_tables():
_ = cur.execute("DROP TABLE IF EXISTS participation")
_ = cur.execute(
"""
CREATE TABLE participation (
div_num integer,
div_name text,
sch_num integer,
sch_name text,
sch_type text,
low_grade text,
high_grade text,
subject text,
subgroup text,
number_tested integer,
number_students integer,
part_rate_2324 real
)
"""
)
_ = cur.execute("DROP TABLE IF EXISTS test_results")
_ = cur.execute(
"""
CREATE TABLE test_results (
div_num integer,
div_name text,
sch_num integer,
sch_name text,
sch_type text,
low_grade text,
high_grade text,
subject text,
grade_level text,
test_name text,
pass_rate_2122 integer,
pass_rate_2223 integer,
pass_rate_2324 integer,
adv_rate_2122 integer,
adv_rate_2223 integer,
adv_rate_2324 integer
)"""
)
con.commit()
def import_data():
with open("School-Participation_rates_23_24.csv", newline="") as file:
reader = csv.reader(file)
_ = next(reader) # skip header row
for row in reader:
_ = cur.execute(
"""INSERT INTO participation
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
row,
)
with open("School_Test_by_level_2023_2024.csv", newline="") as file:
reader = csv.reader(file)
_ = next(reader) # skip header row
for row in reader:
_ = cur.execute(
"""INSERT INTO test_results
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
row,
)
con.commit()
def query_school(div_num: int, sch_num: int):
# Query the participation rates
res = cur.execute(
"""
SELECT div_name, sch_name, subject, number_tested, number_students
FROM participation
WHERE div_num = ?
AND sch_num = ?
AND subgroup = 'All Students'
""",
(div_num, sch_num),
)
students_tested: list[tuple[str, str, str, int, int]] = res.fetchall()
print()
print(f"Results for {students_tested[0][1]}, {students_tested[0][0]}")
print()
for row in students_tested:
print(f"Subject: {row[2]}, Tested: {row[3]}/{row[4]}")
res = cur.execute(
"""
SELECT subject, grade_level, test_name, pass_rate_2324
FROM test_results
WHERE div_num = ?
AND sch_num = ?
""",
(div_num, sch_num),
)
data: list[tuple[str, int, str, float | str | None]] = res.fetchall()
print()
for row in data:
if row[3] is not None and row[3] != "<":
print(f"{row[0]}, {row[1]}, {row[2]}, 2023-24 Pass Rate: {row[3]}")
print()
def update_school(div_num: int, sch_num: int):
print("Hacking database...", end="")
_ = cur.execute(
"""
UPDATE test_results
SET pass_rate_2324 = 100, adv_rate_2324 = 100
WHERE div_num = ?
AND sch_num = ?
""",
(div_num, sch_num),
)
con.commit()
print("success!\nHave a nice day.")
# DO NOT EDIT THESE VARIABLES; use them in your functions.
# Connect to the SQLite database file and create a cursor.
if __name__ == "__main__":
con = sqlite3.connect("vdoe.db")
cur = con.cursor()
create_tables()
import_data()
div_num = int(input("Enter division number: "))
sch_num = int(input("Enter school number: "))
query_school(div_num, sch_num)
update_school(div_num, sch_num)