Code is posted after explanation. Due to the size of the project, this is being posted in three separate posts. This also ensures each post is more focused. * Post 1 of 3, Core: https://codereview.stackexchange.com/q/276927/257004 * Post 2 of 3, CLI: https://codereview.stackexchange.com/q/276928/257004 This is a follow-up to an [earlier version of the same project](https://codereview.stackexchange.com/questions/276225/newspaper-bill-calculator-cli-with-python-1-of-3-core). The feedback from the last round is tracked in an [issue](https://github.com/eccentricOrange/npbc/issues/20). ___ ## What is this? This application helps you calculate monthly newspaper bills. The goal is to generate a message that I can paste into WhatsApp and send to my newspaper vendor. The end result here is a CLI tool that will be later used as a back-end to build GUIs (hence learn about: C\#, HTML/CSS/JS, Flutter). In its current form, everything will be "compiled" by PyInstaller into one-file stand-alone executables for the end-user using GitHub Actions. The other important goal was to be a testbed for learning a bunch of new tools: more Python libraries, SQL connectors, GitHub Actions (CI/CD, if I understand correctly), unit tests, CLI libraries, type-hinting, regex. I had [earlier built](https://github.com/eccentricOrange/Newspaper-Bill-Calculator-v2) this on a different platform, so I now have a solid idea of how this application is used. ## Key concepts 1. Each newspaper has a certain cost per day of the week 2. Each newspaper may or may not be delivered on a given day 3. Each newspaper has a name, and a number called a key 4. You may register any dates when you didn't receive a paper in advance using the `addudl` command 5. Once you calculate, the results are displayed and logged. ## What files exist? (ignoring conventional ones like `README` and `requirements.txt`) | File | Purpose/Description | Review | | -- | -- | -- | | `npbc_core.py` | Provide the core functionality: the calculation, parsing and validation of user input, interaction with the DB etc. Later on, some functionality from this will be extracted to create server-side code that can service more users, but I have to learn a lot more before getting there. | Please review this. | | `npbc_regex.py` | Contains all the regex statements used to validate and parse user input. | Please review this. | | `npbc_exceptions.py` | Defines classes for all the custom exceptions used by the core and the CLI. | Please review this. | | `npbc_cli.py` | Import functionality from `npbc_core.py` and wrap a CLI layer on it using `argparse`. Also provide some additional validation. | Please review this. | | `npbc_updater.py` | Provide a utility to update the application on the user's end. | Don't bother reviewing this (code not included). | | `test_core.py` | Test the functionality of the core file (pytest), except anything to do with the database. | Please review this. | | `test_db.py` | Test the functionality of the core file (pytest), for anything to do with the database. | Please review this. | | `test_regex.py` | Test the functionality of the regex statements. | Please review this. | | `data/schema.sql` | Database schema. In my local environment, the `data` folder also has a test database file (but I don't want to upload this online). | Please review this if you can (not high priority). | | `data/test.sql` | SQL statements to generate test data for `test_db.py`. | Please review this if you can (not high priority). | ___ # `data/schema.sql` ```sql CREATE TABLE IF NOT EXISTS papers ( paper_id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, CONSTRAINT unique_paper_name UNIQUE (name) ); CREATE TABLE IF NOT EXISTS cost_and_delivery_data ( paper_day_id INTEGER PRIMARY KEY AUTOINCREMENT, paper_id INTEGER NOT NULL REFERENCES papers(paper_id), day_id INTEGER NOT NULL, cost REAL NOT NULL, delivered INTEGER NOT NULL, CONSTRAINT unique_paper_day UNIQUE (paper_id, day_id) ); CREATE TABLE IF NOT EXISTS undelivered_strings ( string_id INTEGER PRIMARY KEY AUTOINCREMENT, year INTEGER NOT NULL CHECK (year >= 0), month INTEGER NOT NULL CHECK (month >= 0 AND month <= 12), paper_id INTEGER NOT NULL REFERENCES papers(paper_id), string TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS logs ( log_id INTEGER PRIMARY KEY AUTOINCREMENT, paper_id INTEGER NOT NULL REFERENCES papers(paper_id), timestamp TEXT NOT NULL, month INTEGER NOT NULL CHECK (month >= 0 AND month <= 12), year INTEGER NOT NULL CHECK (year >= 0), CONSTRAINT unique_log UNIQUE (timestamp, paper_id, month, year) ); CREATE TABLE IF NOT EXISTS undelivered_dates_logs ( undelivered_dates_log_id INTEGER PRIMARY KEY AUTOINCREMENT, log_id INTEGER NOT NULL REFERENCES logs(log_id), date_not_delivered TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS cost_logs ( cost_log_id INTEGER PRIMARY KEY AUTOINCREMENT, log_id INTEGER NOT NULL REFERENCES logs(log_id), cost REAL NOT NULL ); ``` # `data/test.sql` ```sql CREATE TABLE IF NOT EXISTS papers ( paper_id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, CONSTRAINT unique_paper_name UNIQUE (name) ); CREATE TABLE IF NOT EXISTS cost_and_delivery_data ( paper_day_id INTEGER PRIMARY KEY AUTOINCREMENT, paper_id INTEGER NOT NULL REFERENCES papers(paper_id), day_id INTEGER NOT NULL, cost REAL NOT NULL, delivered INTEGER NOT NULL, CONSTRAINT unique_paper_day UNIQUE (paper_id, day_id) ); CREATE TABLE IF NOT EXISTS undelivered_strings ( string_id INTEGER PRIMARY KEY AUTOINCREMENT, year INTEGER NOT NULL CHECK (year >= 0), month INTEGER NOT NULL CHECK (month >= 0 AND month <= 12), paper_id INTEGER NOT NULL REFERENCES papers(paper_id), string TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS logs ( log_id INTEGER PRIMARY KEY AUTOINCREMENT, paper_id INTEGER NOT NULL REFERENCES papers(paper_id), timestamp TEXT NOT NULL, month INTEGER NOT NULL CHECK (month >= 0 AND month <= 12), year INTEGER NOT NULL CHECK (year >= 0), CONSTRAINT unique_log UNIQUE (timestamp, paper_id, month, year) ); CREATE TABLE IF NOT EXISTS undelivered_dates_logs ( undelivered_dates_log_id INTEGER PRIMARY KEY AUTOINCREMENT, log_id INTEGER NOT NULL REFERENCES logs(log_id), date_not_delivered TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS cost_logs ( cost_log_id INTEGER PRIMARY KEY AUTOINCREMENT, log_id INTEGER NOT NULL REFERENCES logs(log_id), cost REAL NOT NULL ); ``` # `test_db.py` ```python """ test data-dependent functions from the core - all of these depend on the DB - for consistency, the DB will always be initialised with the same data - the test data is contained in `data/test.sql` - the schema is the same as the core (`data/schema.sql` during development) """ from datetime import date, datetime from pathlib import Path from sqlite3 import connect from typing import Counter from pytest import raises import npbc_cli import npbc_core import npbc_exceptions ACTIVE_DIRECTORY = Path("data") DATABASE_PATH = ACTIVE_DIRECTORY / "npbc.db" SCHEMA_PATH = ACTIVE_DIRECTORY / "schema.sql" TEST_SQL = ACTIVE_DIRECTORY / "test.sql" def setup_db(): DATABASE_PATH.unlink(missing_ok=True) with connect(DATABASE_PATH) as connection: connection.executescript(SCHEMA_PATH.read_text()) connection.commit() connection.executescript(TEST_SQL.read_text()) connection.close() def test_db_creation(): DATABASE_PATH.unlink(missing_ok=True) assert not DATABASE_PATH.exists() try: npbc_cli.main([]) except SystemExit: pass assert DATABASE_PATH.exists() def test_get_papers(): setup_db() known_data = [ (1, 'paper1', 0, 0, 0), (1, 'paper1', 1, 1, 6.4), (1, 'paper1', 2, 0, 0), (1, 'paper1', 3, 0, 0), (1, 'paper1', 4, 0, 0), (1, 'paper1', 5, 1, 7.9), (1, 'paper1', 6, 1, 4), (2, 'paper2', 0, 0, 0), (2, 'paper2', 1, 0, 0), (2, 'paper2', 2, 0, 0), (2, 'paper2', 3, 0, 0), (2, 'paper2', 4, 1, 3.4), (2, 'paper2', 5, 0, 0), (2, 'paper2', 6, 1, 8.4), (3, 'paper3', 0, 1, 2.4), (3, 'paper3', 1, 1, 4.6), (3, 'paper3', 2, 0, 0), (3, 'paper3', 3, 0, 0), (3, 'paper3', 4, 1, 3.4), (3, 'paper3', 5, 1, 4.6), (3, 'paper3', 6, 1, 6) ] assert Counter(npbc_core.get_papers()) == Counter(known_data) def test_get_undelivered_strings(): setup_db() known_data = [ (1, 1, 2020, 11, '5'), (2, 1, 2020, 11, '6-12'), (3, 2, 2020, 11, 'sundays'), (4, 3, 2020, 11, '2-tuesday'), (5, 3, 2020, 10, 'all') ] assert Counter(npbc_core.get_undelivered_strings()) == Counter(known_data) assert Counter(npbc_core.get_undelivered_strings(string_id=3)) == Counter([known_data[2]]) assert Counter(npbc_core.get_undelivered_strings(month=11)) == Counter(known_data[:4]) assert Counter(npbc_core.get_undelivered_strings(paper_id=1)) == Counter(known_data[:2]) assert Counter(npbc_core.get_undelivered_strings(paper_id=1, string='6-12')) == Counter([known_data[1]]) with raises(npbc_exceptions.StringNotExists): npbc_core.get_undelivered_strings(year=1986) def test_delete_paper(): setup_db() npbc_core.delete_existing_paper(2) known_data = [ (1, 'paper1', 0, 0, 0), (1, 'paper1', 1, 1, 6.4), (1, 'paper1', 2, 0, 0), (1, 'paper1', 3, 0, 0), (1, 'paper1', 4, 0, 0), (1, 'paper1', 5, 1, 7.9), (1, 'paper1', 6, 1, 4), (3, 'paper3', 0, 1, 2.4), (3, 'paper3', 1, 1, 4.6), (3, 'paper3', 2, 0, 0), (3, 'paper3', 3, 0, 0), (3, 'paper3', 4, 1, 3.4), (3, 'paper3', 5, 1, 4.6), (3, 'paper3', 6, 1, 6) ] assert Counter(npbc_core.get_papers()) == Counter(known_data) with raises(npbc_exceptions.PaperNotExists): npbc_core.delete_existing_paper(7) npbc_core.delete_existing_paper(2) def test_add_paper(): setup_db() known_data = [ (1, 'paper1', 0, 0, 0), (1, 'paper1', 1, 1, 6.4), (1, 'paper1', 2, 0, 0), (1, 'paper1', 3, 0, 0), (1, 'paper1', 4, 0, 0), (1, 'paper1', 5, 1, 7.9), (1, 'paper1', 6, 1, 4), (2, 'paper2', 0, 0, 0), (2, 'paper2', 1, 0, 0), (2, 'paper2', 2, 0, 0), (2, 'paper2', 3, 0, 0), (2, 'paper2', 4, 1, 3.4), (2, 'paper2', 5, 0, 0), (2, 'paper2', 6, 1, 8.4), (3, 'paper3', 0, 1, 2.4), (3, 'paper3', 1, 1, 4.6), (3, 'paper3', 2, 0, 0), (3, 'paper3', 3, 0, 0), (3, 'paper3', 4, 1, 3.4), (3, 'paper3', 5, 1, 4.6), (3, 'paper3', 6, 1, 6), (4, 'paper4', 0, 1, 4), (4, 'paper4', 1, 0, 0), (4, 'paper4', 2, 1, 2.6), (4, 'paper4', 3, 0, 0), (4, 'paper4', 4, 0, 0), (4, 'paper4', 5, 1, 1), (4, 'paper4', 6, 1, 7) ] npbc_core.add_new_paper( 'paper4', [True, False, True, False, False, True, True], [4, 0, 2.6, 0, 0, 1, 7] ) assert Counter(npbc_core.get_papers()) == Counter(known_data) with raises(npbc_exceptions.PaperAlreadyExists): npbc_core.add_new_paper( 'paper4', [True, False, True, False, False, True, True], [4, 0, 2.6, 0, 0, 1, 7] ) def test_edit_paper(): setup_db() known_data = [ (1, 'paper1', 0, 0, 0), (1, 'paper1', 1, 1, 6.4), (1, 'paper1', 2, 0, 0), (1, 'paper1', 3, 0, 0), (1, 'paper1', 4, 0, 0), (1, 'paper1', 5, 1, 7.9), (1, 'paper1', 6, 1, 4), (2, 'paper2', 0, 0, 0), (2, 'paper2', 1, 0, 0), (2, 'paper2', 2, 0, 0), (2, 'paper2', 3, 0, 0), (2, 'paper2', 4, 1, 3.4), (2, 'paper2', 5, 0, 0), (2, 'paper2', 6, 1, 8.4), (3, 'paper3', 0, 1, 2.4), (3, 'paper3', 1, 1, 4.6), (3, 'paper3', 2, 0, 0), (3, 'paper3', 3, 0, 0), (3, 'paper3', 4, 1, 3.4), (3, 'paper3', 5, 1, 4.6), (3, 'paper3', 6, 1, 6) ] npbc_core.edit_existing_paper( 1, days_delivered=[True, False, True, False, False, True, True], days_cost=[6.4, 0, 0, 0, 0, 7.9, 4] ) known_data[0] = (1, 'paper1', 0, 1, 6.4) known_data[1] = (1, 'paper1', 1, 0, 0) known_data[2] = (1, 'paper1', 2, 1, 0) known_data[3] = (1, 'paper1', 3, 0, 0) known_data[4] = (1, 'paper1', 4, 0, 0) known_data[5] = (1, 'paper1', 5, 1, 7.9) known_data[6] = (1, 'paper1', 6, 1, 4) assert Counter(npbc_core.get_papers()) == Counter(known_data) npbc_core.edit_existing_paper( 3, name="New paper" ) known_data[14] = (3, 'New paper', 0, 1, 2.4) known_data[15] = (3, 'New paper', 1, 1, 4.6) known_data[16] = (3, 'New paper', 2, 0, 0) known_data[17] = (3, 'New paper', 3, 0, 0) known_data[18] = (3, 'New paper', 4, 1, 3.4) known_data[19] = (3, 'New paper', 5, 1, 4.6) known_data[20] = (3, 'New paper', 6, 1, 6) assert Counter(npbc_core.get_papers()) == Counter(known_data) with raises(npbc_exceptions.PaperNotExists): npbc_core.edit_existing_paper(7, name="New paper") def test_delete_string(): known_data = [ (1, 1, 2020, 11, '5'), (2, 1, 2020, 11, '6-12'), (3, 2, 2020, 11, 'sundays'), (4, 3, 2020, 11, '2-tuesday'), (5, 3, 2020, 10, 'all') ] setup_db() npbc_core.delete_undelivered_string(string='all') assert Counter(npbc_core.get_undelivered_strings()) == Counter(known_data[:4]) setup_db() npbc_core.delete_undelivered_string(month=11) assert Counter(npbc_core.get_undelivered_strings()) == Counter([known_data[4]]) setup_db() npbc_core.delete_undelivered_string(paper_id=1) assert Counter(npbc_core.get_undelivered_strings()) == Counter(known_data[2:]) setup_db() with raises(npbc_exceptions.StringNotExists): npbc_core.delete_undelivered_string(string='not exists') with raises(npbc_exceptions.NoParameters): npbc_core.delete_undelivered_string() def test_add_string(): setup_db() known_data = [ (1, 1, 2020, 11, '5'), (2, 1, 2020, 11, '6-12'), (3, 2, 2020, 11, 'sundays'), (4, 3, 2020, 11, '2-tuesday'), (5, 3, 2020, 10, 'all') ] npbc_core.add_undelivered_string(4, 2017, 3, 'sundays') known_data.append((6, 3, 2017, 4, 'sundays')) assert Counter(npbc_core.get_undelivered_strings()) == Counter(known_data) npbc_core.add_undelivered_string(9, 2017, None, '11') known_data.append((7, 1, 2017, 9, '11')) known_data.append((8, 2, 2017, 9, '11')) known_data.append((9, 3, 2017, 9, '11')) assert Counter(npbc_core.get_undelivered_strings()) == Counter(known_data) def test_save_results(): setup_db() known_data = [ (1, 1, 2020, '04/01/2022 01:05:42 AM', '2020-01-01'), (1, 1, 2020, '04/01/2022 01:05:42 AM', '2020-01-02'), (2, 1, 2020, '04/01/2022 01:05:42 AM', '2020-01-01'), (2, 1, 2020, '04/01/2022 01:05:42 AM', '2020-01-05'), (2, 1, 2020, '04/01/2022 01:05:42 AM', '2020-01-03'), (1, 1, 2020, '04/01/2022 01:05:42 AM', 105.0), (2, 1, 2020, '04/01/2022 01:05:42 AM', 51.0), (3, 1, 2020, '04/01/2022 01:05:42 AM', 647.0) ] npbc_core.save_results( {1: 105, 2: 51, 3: 647}, { 1: set([date(month=1, day=1, year=2020), date(month=1, day=2, year=2020)]), 2: set([date(month=1, day=1, year=2020), date(month=1, day=5, year=2020), date(month=1, day=3, year=2020)]), 3: set() }, 1, 2020, datetime(year=2022, month=1, day=4, hour=1, minute=5, second=42) ) assert Counter(npbc_core.get_logged_data()) == Counter(known_data) ``` # All code that uses the DB in `npbc_core.py` ### Setup ```python ## paths for the folder containing schema and database files # during normal use, the DB will be in ~/.npbc (where ~ is the user's home directory) and the schema will be bundled with the executable # during development, the DB and schema will both be in "data" # default to PRODUCTION DATABASE_DIR = Path.home() / '.npbc' SCHEMA_PATH = Path(__file__).parent / 'schema.sql' # if in a development environment, set the paths to the data folder if environ.get('NPBC_DEVELOPMENT') or environ.get('CI'): DATABASE_DIR = Path('data') SCHEMA_PATH = Path('data') / 'schema.sql' DATABASE_PATH = DATABASE_DIR / 'npbc.db' ## list constant for names of weekdays WEEKDAY_NAMES = list(weekday_names_iterable) def setup_and_connect_DB() -> None: """ensure DB exists and it's set up with the schema""" DATABASE_DIR.mkdir(parents=True, exist_ok=True) DATABASE_PATH.touch(exist_ok=True) with connect(DATABASE_PATH) as connection: connection.executescript(SCHEMA_PATH.read_text()) connection.close() ``` ### Fetch cost and delivery data for calculation ```python def get_cost_and_delivery_data(paper_id: int, connection: Connection) -> list[tuple[bool, float]]: """get the cost and delivery data for a given paper from the DB""" query = """ SELECT delivered, cost FROM cost_and_delivery_data WHERE paper_id = ? ORDER BY day_id; """ # return a list but convert the delivery data to Booleans because SQLite won't do it return list(map( lambda row: (bool(row[0]), row[1]), connection.execute(query, (paper_id,)).fetchall() )) ``` ### Calculate the cost ```python def calculate_cost_of_all_papers(undelivered_strings: dict[int, list[str]], month: int, year: int) -> tuple[ dict[int, float], float, dict[int, set[date_type]] ]: """calculate the cost of all papers for the full month - return data about the cost of each paper, the total cost, and dates when each paper was not delivered""" NUMBER_OF_EACH_WEEKDAY = list(get_number_of_each_weekday(month, year)) cost_and_delivery_data = {} # get the IDs of papers that exist with connect(DATABASE_PATH) as connection: papers = connection.execute("SELECT paper_id FROM papers;").fetchall() # get the data about cost and delivery for each paper cost_and_delivery_data = [ get_cost_and_delivery_data(paper_id, connection) for paper_id, in papers # type: ignore ] connection.close() # initialize a "blank" dictionary that will eventually contain any dates when a paper was not delivered undelivered_dates: dict[int, set[date_type]] = { int(paper_id): set() for paper_id, in papers # type: ignore } # calculate the undelivered dates for each paper for paper_id, strings in undelivered_strings.items(): undelivered_dates[paper_id].update( parse_undelivered_strings(month, year, *strings) ) # calculate the cost of each paper costs = { paper_id: calculate_cost_of_one_paper( NUMBER_OF_EACH_WEEKDAY, undelivered_dates[paper_id], cost_and_delivery_data[index] ) for index, (paper_id,) in enumerate(papers) # type: ignore } # calculate the total cost of all papers total = sum(costs.values()) return costs, total, undelivered_dates ``` ### Format user output and save data to logs ```python def save_results( costs: dict[int, float], undelivered_dates: dict[int, set[date_type]], month: int, year: int, custom_timestamp: datetime | None = None ) -> None: """save the results of undelivered dates to the DB - save the dates any paper was not delivered - save the final cost of each paper""" timestamp = (custom_timestamp if custom_timestamp else datetime.now()).strftime(r'%d/%m/%Y %I:%M:%S %p') with connect(DATABASE_PATH) as connection: # create log entries for each paper log_ids = { paper_id: connection.execute( """ INSERT INTO logs (paper_id, month, year, timestamp) VALUES (?, ?, ?, ?) RETURNING logs.log_id; """, (paper_id, month, year, timestamp) ).fetchone()[0] for paper_id in costs.keys() } # create cost entries for each paper for paper_id, log_id in log_ids.items(): connection.execute( """ INSERT INTO cost_logs (log_id, cost) VALUES (?, ?); """, (log_id, costs[paper_id]) ) # create undelivered date entries for each paper for paper_id, dates in undelivered_dates.items(): for date in dates: connection.execute( """ INSERT INTO undelivered_dates_logs (log_id, date_not_delivered) VALUES (?, ?); """, (log_ids[paper_id], date.strftime("%Y-%m-%d")) ) connection.close() def format_output(costs: dict[int, float], total: float, month: int, year: int) -> Generator[str, None, None]: """format the output of calculating the cost of all papers""" # output the name of the month for which the total cost was calculated yield f"For {date_type(year=year, month=month, day=1).strftime(r'%B %Y')},\n" # output the total cost of all papers yield f"*TOTAL*: {total:.2f}" # output the cost of each paper with its name with connect(DATABASE_PATH) as connection: papers = dict(connection.execute("SELECT paper_id, name FROM papers;").fetchall()) for paper_id, cost in costs.items(): yield f"{papers[paper_id]}: {cost:.2f}" connection.close() ``` ### Add, edit, or delete a paper ```python def add_new_paper(name: str, days_delivered: list[bool], days_cost: list[float]) -> None: """add a new paper - do not allow if the paper already exists""" with connect(DATABASE_PATH) as connection: # check if the paper already exists if connection.execute( "SELECT EXISTS (SELECT 1 FROM papers WHERE name = ?);", (name,)).fetchone()[0]: raise npbc_exceptions.PaperAlreadyExists(f"Paper \"{name}\" already exists." ) # insert the paper paper_id = connection.execute( "INSERT INTO papers (name) VALUES (?) RETURNING papers.paper_id;", (name,) ).fetchone()[0] # create cost and delivered entries for each day for day_id, (delivered, cost) in enumerate(zip(days_delivered, days_cost)): connection.execute( "INSERT INTO cost_and_delivery_data (paper_id, day_id, delivered, cost) VALUES (?, ?, ?, ?);", (paper_id, day_id, delivered, cost) ) connection.close() def edit_existing_paper( paper_id: int, name: str | None = None, days_delivered: list[bool] | None = None, days_cost: list[float] | None = None ) -> None: """edit an existing paper do not allow if the paper does not exist""" with connect(DATABASE_PATH) as connection: # check if the paper exists if not connection.execute( "SELECT EXISTS (SELECT 1 FROM papers WHERE paper_id = ?);", (paper_id,)).fetchone()[0]: raise npbc_exceptions.PaperNotExists(f"Paper with ID {paper_id} does not exist." ) # update the paper name if name is not None: connection.execute( "UPDATE papers SET name = ? WHERE paper_id = ?;", (name, paper_id) ) # update the costs of each day if days_cost is not None: for day_id, cost in enumerate(days_cost): connection.execute( "UPDATE cost_and_delivery_data SET cost = ? WHERE paper_id = ? AND day_id = ?;", (cost, paper_id, day_id) ) # update the delivered status of each day if days_delivered is not None: for day_id, delivered in enumerate(days_delivered): connection.execute( "UPDATE cost_and_delivery_data SET delivered = ? WHERE paper_id = ? AND day_id = ?;", (delivered, paper_id, day_id) ) connection.close() def delete_existing_paper(paper_id: int) -> None: """delete an existing paper - do not allow if the paper does not exist""" with connect(DATABASE_PATH) as connection: # check if the paper exists if not connection.execute( "SELECT EXISTS (SELECT 1 FROM papers WHERE paper_id = ?);", (paper_id,)).fetchone()[0]: raise npbc_exceptions.PaperNotExists(f"Paper with ID {paper_id} does not exist." ) # delete the paper connection.execute( "DELETE FROM papers WHERE paper_id = ?;", (paper_id,) ) # delete the costs and delivery data for the paper connection.execute( "DELETE FROM cost_and_delivery_data WHERE paper_id = ?;", (paper_id,) ) connection.close() ``` ### Add or delete undelivered strings ```python def add_undelivered_string(month: int, year: int, paper_id: int | None = None, *undelivered_strings: str) -> None: """record strings for date(s) paper(s) were not delivered - if no paper ID is specified, all papers are assumed""" # validate the strings validate_undelivered_string(*undelivered_strings) # if a paper ID is given if paper_id: # check that specified paper exists in the database with connect(DATABASE_PATH) as connection: if not connection.execute( "SELECT EXISTS (SELECT 1 FROM papers WHERE paper_id = ?);", (paper_id,)).fetchone()[0]: raise npbc_exceptions.PaperNotExists(f"Paper with ID {paper_id} does not exist." ) # add the string(s) params = [ (month, year, paper_id, string) for string in undelivered_strings ] connection.executemany("INSERT INTO undelivered_strings (month, year, paper_id, string) VALUES (?, ?, ?, ?);", params) connection.close() # if no paper ID is given else: # get the IDs of all papers with connect(DATABASE_PATH) as connection: paper_ids = [ row[0] for row in connection.execute( "SELECT paper_id FROM papers;" ) ] # add the string(s) params = [ (month, year, paper_id, string) for paper_id in paper_ids for string in undelivered_strings ] connection.executemany("INSERT INTO undelivered_strings (month, year, paper_id, string) VALUES (?, ?, ?, ?);", params) connection.close() def delete_undelivered_string( string_id: int | None = None, string: str | None = None, paper_id: int | None = None, month: int | None = None, year: int | None = None ) -> None: """delete an existing undelivered string - do not allow if the string does not exist""" # initialize parameters for the WHERE clause of the SQL query parameters = [] values = [] # check each parameter and add it to the WHERE clause if it is given if string_id: parameters.append("string_id") values.append(string_id) if string: parameters.append("string") values.append(string) if paper_id: parameters.append("paper_id") values.append(paper_id) if month: parameters.append("month") values.append(month) if year: parameters.append("year") values.append(year) # if no parameters are given, raise an error if not parameters: raise npbc_exceptions.NoParameters("No parameters given.") with connect(DATABASE_PATH) as connection: # check if the string exists check_query = "SELECT EXISTS (SELECT 1 FROM undelivered_strings" conditions = ' AND '.join( f"{parameter} = ?" for parameter in parameters ) if (1,) not in connection.execute(f"{check_query} WHERE {conditions});", values).fetchall(): raise npbc_exceptions.StringNotExists("String with given parameters does not exist.") # if the string did exist, delete it delete_query = "DELETE FROM undelivered_strings" connection.execute(f"{delete_query} WHERE {conditions};", values) connection.close() ``` ### Get papers, undelivered strings, or logged data ```python def get_papers() -> list[tuple[int, str, int, int, float]]: """get all papers - returns a list of tuples containing the following fields: paper_id, paper_name, day_id, paper_delivered, paper_cost""" raw_data = [] query = """ SELECT papers.paper_id, papers.name, cost_and_delivery_data.day_id, cost_and_delivery_data.delivered, cost_and_delivery_data.cost FROM papers INNER JOIN cost_and_delivery_data ON papers.paper_id = cost_and_delivery_data.paper_id ORDER BY papers.paper_id, cost_and_delivery_data.day_id; """ with connect(DATABASE_PATH) as connection: raw_data = connection.execute(query).fetchall() connection.close() return raw_data def get_undelivered_strings( string_id: int | None = None, month: int | None = None, year: int | None = None, paper_id: int | None = None, string: str | None = None ) -> list[tuple[int, int, int, int, str]]: """get undelivered strings - the user may specify as many as they want parameters - available parameters: string_id, month, year, paper_id, string - returns a list of tuples containing the following fields: string_id, paper_id, year, month, string""" # initialize parameters for the WHERE clause of the SQL query parameters = [] values = [] data = [] # check each parameter and add it to the WHERE clause if it is given if string_id: parameters.append("string_id") values.append(string_id) if month: parameters.append("month") values.append(month) if year: parameters.append("year") values.append(year) if paper_id: parameters.append("paper_id") values.append(paper_id) if string: parameters.append("string") values.append(string) with connect(DATABASE_PATH) as connection: # generate the SQL query main_query = "SELECT string_id, paper_id, year, month, string FROM undelivered_strings" if not parameters: query = f"{main_query};" else: conditions = ' AND '.join( f"{parameter} = ?" for parameter in parameters ) query = f"{main_query} WHERE {conditions};" data = connection.execute(query, values).fetchall() connection.close() # if no data was found, raise an error if not data: raise npbc_exceptions.StringNotExists("String with given parameters does not exist.") return data def get_logged_data( query_paper_id: int | None = None, query_log_id: int | None = None, query_month: int | None = None, query_year: int | None = None, query_timestamp: date_type | None = None ) -> Generator[tuple[int, int, int, int, str, str | float], None, None]: """get logged data - the user may specify as parameters many as they want - available parameters: paper_id, log_id, month, year, timestamp - yields: tuples containing the following fields: log_id, paper_id, month, year, timestamp, date | cost.""" # initialize parameters for the WHERE clause of the SQL query parameters = [] values = () # check each parameter and add it to the WHERE clause if it is given if query_paper_id: parameters.append("paper_id") values += (query_paper_id,) if query_log_id: parameters.append("log_id") values += (query_log_id,) if query_month: parameters.append("month") values += (query_month,) if query_year: parameters.append("year") values += (query_year,) if query_timestamp: parameters.append("timestamp") values += (query_timestamp.strftime(r'%d/%m/%Y %I:%M:%S %p'),) # generate the SQL query logs_base_query = """ SELECT log_id, paper_id, timestamp, month, year FROM logs ORDER BY log_id, paper_id """ if parameters: conditions = ' AND '.join( f"{parameter} = ?" for parameter in parameters ) logs_query = f"{logs_base_query} WHERE {conditions};" else: logs_query = f"{logs_base_query};" dates_query = "SELECT log_id, date_not_delivered FROM undelivered_dates_logs;" costs_query = "SELECT log_id, cost FROM cost_logs;" with connect(DATABASE_PATH) as connection: logs = { log_id: [paper_id, month, year, timestamp] for log_id, paper_id, timestamp, month, year in connection.execute(logs_query, values).fetchall() } dates = connection.execute(dates_query).fetchall() costs = connection.execute(costs_query).fetchall() for log_id, date in dates: yield tuple(logs[log_id] + [date]) for log_id, cost in costs: yield tuple(logs[log_id] + [float(cost)]) connection.close() ``` ___ If you need it, here is a link to the GitHub repo for this project. It's at the same commit as the code above, and I won't edit this so that any discussion is consistent. https://github.com/eccentricOrange/npbc/tree/041a949bbba018531ec590a2193523f1530659aa