UPDATE on Newspaper Bill Calculator CLI with Python (3 of 3, Database)

Question

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: UPDATE 1 on Newspaper Bill Calculator CLI with Python (1 of 3, Core)
• Post 2 of 3, CLI: UPDATE 1 on Newspaper Bill Calculator CLI with Python (2 of 3, CLI)

This is a follow-up to an earlier version of the same project. The feedback from the last round is tracked in an issue.

---

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 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

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

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

"""
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

## 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

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

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

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

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

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

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