3
\$\begingroup\$

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.


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 copied to your clipboard

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_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). This isn't as exhaustive as I'd like, but it did a good job of capturing many of my mistakes. 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).

Known problems

  • Tests are not exhaustive (please suggest anything you think of).
  • Tests are not well commented (working on this right now in a local branch).
  • SQL injection is possible in some cases by -k/--key CLI parameters, if you can figure out a way to insert a semicolon in an integer. I will remove this in a future version, once I find a way to improve or remove the generate_sql_query() function.
  • A lot of documentation is tied up in the CLI UI and comments, and is not an explicit document.

data/schema.sql (SQLite 3)

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 papers_days_delivered (
    paper_id INTEGER NOT NULL,
    day_id INTEGER NOT NULL,
    delivered INTEGER NOT NULL,
    FOREIGN KEY(paper_id) REFERENCES papers(paper_id),
    CONSTRAINT unique_paper_day UNIQUE (paper_id, day_id)
);
CREATE TABLE IF NOT EXISTS papers_days_cost(
    paper_id INTEGER NOT NULL,
    day_id INTEGER NOT NULL,
    cost INTEGER,
    FOREIGN KEY(paper_id) REFERENCES papers(paper_id),
    CONSTRAINT unique_paper_day UNIQUE (paper_id, day_id)
);
CREATE TABLE IF NOT EXISTS undelivered_strings (
    entry_id INTEGER PRIMARY KEY AUTOINCREMENT,
    year INTEGER NOT NULL,
    month INTEGER NOT NULL,
    paper_id INTEGER NOT NULL,
    string TEXT NOT NULL,
    FOREIGN KEY (paper_id) REFERENCES papers(paper_id)
);
CREATE TABLE IF NOT EXISTS undelivered_dates (
    entry_id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp TEXT NOT NULL,
    year INTEGER NOT NULL,
    month INTEGER NOT NULL,
    paper_id INTEGER NOT NULL,
    dates TEXT NOT NULL,
    FOREIGN KEY (paper_id) REFERENCES papers(paper_id)
);
CREATE INDEX IF NOT EXISTS search_strings ON undelivered_strings(year, month);
CREATE INDEX IF NOT EXISTS paper_names ON papers(name);

All code where a query/call to DB is made

This is provided for context, and is duplicated from the other posts.

npbc_core.py

Setup

from sqlite3 import connect
from pathlib import Path

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

DATABASE_DIR = Path().home() / '.npbc'  # normal use path
# DATABASE_DIR = Path('data')  # development path

DATABASE_PATH = DATABASE_DIR / 'npbc.db'

SCHEMA_PATH = Path(__file__).parent / 'schema.sql'  # normal use path
# SCHEMA_PATH = DATABASE_DIR / 'schema.sql'  # development path

## ensure DB exists and it's set up with the schema
def setup_and_connect_DB() -> None:
    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.commit()

Query generator

## generate a "SELECT" SQL query
 # use params to specify columns to select, and "WHERE" conditions
def generate_sql_query(table_name: str, conditions: dict[str, int | str] | None = None, columns: list[str] | None = None) -> str:
    sql_query = f"SELECT"
    
    if columns:
        sql_query += f" {', '.join(columns)}"
    
    else:
        sql_query += f" *"
    
    sql_query += f" FROM {table_name}"

    if conditions:
        conditions_segment = ' AND '.join([
            f"{parameter_name} = {parameter_value}"
            for parameter_name, parameter_value in conditions.items()
        ])
        
        sql_query += f" WHERE {conditions_segment}"

    return f"{sql_query};"

Query (call to DB)

## execute a "SELECT" SQL query and return the results
def query_database(query: str) -> list[tuple]:
    with connect(DATABASE_PATH) as connection:
        return connection.execute(query).fetchall()
    
    return []

Fetching data about a given paper, used in calculation

## get the cost and delivery data for a given paper from the DB
 # each of them are converted to a dictionary, whose index is the day_id
 # the two dictionaries are then returned as a tuple
def get_cost_and_delivery_data(paper_id: int) -> tuple[dict[int, float], dict[int, bool]]:
    cost_query = generate_sql_query(
        'papers_days_cost',
        columns=['day_id', 'cost'],
        conditions={'paper_id': paper_id}
    )

    delivery_query = generate_sql_query(
        'papers_days_delivered',
        columns=['day_id', 'delivered'],
        conditions={'paper_id': paper_id}
    )

    with connect(DATABASE_PATH) as connection:
        cost_tuple = connection.execute(cost_query).fetchall()
        delivery_tuple = connection.execute(delivery_query).fetchall()

    cost_dict = {
        day_id: cost
        for day_id, cost in cost_tuple # type: ignore
    }

    delivery_dict = {
        day_id: delivery
        for day_id, delivery in delivery_tuple # type: ignore
    }

    return cost_dict, delivery_dict

Main calculation function

## 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
def calculate_cost_of_all_papers(undelivered_strings: dict[int, str], month: int, year: int) -> tuple[dict[int, float], float, dict[int, set[date_type]]]:
    NUMBER_OF_DAYS_PER_WEEK = get_number_of_days_per_week(month, year)

    # get the IDs of papers that exist
    with connect(DATABASE_PATH) as connection:
        papers = connection.execute(
            generate_sql_query(
                'papers',
                columns=['paper_id']
            )
        ).fetchall()

    # get the data about cost and delivery for each paper
    cost_and_delivery_data = [
        get_cost_and_delivery_data(paper_id)
        for paper_id, in papers # type: ignore
    ]

    # initialize a "blank" dictionary that will eventually contain any dates when a paper was not delivered
    undelivered_dates: dict[int, set[date_type]] = {
        paper_id: {}
        for paper_id, in papers # type: ignore
    }

    # calculate the undelivered dates for each paper
    for paper_id, undelivered_string in undelivered_strings.items(): # type: ignore
        undelivered_dates[paper_id] = parse_undelivered_string(undelivered_string, month, year)

    # calculate the cost of each paper
    costs = {
        paper_id: calculate_cost_of_one_paper(
            NUMBER_OF_DAYS_PER_WEEK,
            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

Save results

## save the results of undelivered dates to the DB
 # save the dates any paper was not delivered
def save_results(undelivered_dates: dict[int, set[date_type]], month: int, year: int) -> None:
    TIMESTAMP = datetime.now().strftime(r'%d/%m/%Y %I:%M:%S %p')

    with connect(DATABASE_PATH) as connection:
        for paper_id, undelivered_date_instances in undelivered_dates.items():
            connection.execute(
                "INSERT INTO undelivered_dates (timestamp, month, year, paper_id, dates) VALUES (?, ?, ?, ?, ?);",
                (
                    TIMESTAMP,
                    month,
                    year,
                    paper_id,
                    ','.join([
                        undelivered_date_instance.strftime(r'%d')
                        for undelivered_date_instance in undelivered_date_instances
                    ])
                )
            )

Add, edit, or delete papers

## add a new paper
 # do not allow if the paper already exists
def add_new_paper(name: str, days_delivered: list[bool], days_cost: list[float]) -> tuple[bool, str]:
    with connect(DATABASE_PATH) as connection:

        # get the names of all papers that already exist
        paper = connection.execute(
            generate_sql_query('papers', columns=['name'], conditions={'name': f"\"{name}\""})
        ).fetchall()

        # if the proposed paper already exists, return an error message
        if paper:
            return False, "Paper already exists. Please try editing the paper instead."
        
        # otherwise, add the paper name to the database
        connection.execute(
            "INSERT INTO papers (name) VALUES (?);",
            (name, )
        )

        # get the ID of the paper that was just added
        paper_id = connection.execute(
            "SELECT paper_id FROM papers WHERE name = ?;",
            (name, )
        ).fetchone()[0]

        # add the cost and delivery data for the paper
        for day_id, (cost, delivered) in enumerate(zip(days_cost, days_delivered)):
            connection.execute(
                "INSERT INTO papers_days_cost (paper_id, day_id, cost) VALUES (?, ?, ?);",
                (paper_id, day_id, cost)
            )
            connection.execute(
                "INSERT INTO papers_days_delivered (paper_id, day_id, delivered) VALUES (?, ?, ?);",
                (paper_id, day_id, delivered)
            )
        
        connection.commit()

        return True, f"Paper {name} added."

    return False, "Something went wrong."


## edit an existing paper
 # do not allow if the paper does not exist
def edit_existing_paper(paper_id: int, name: str | None = None, days_delivered: list[bool] | None = None, days_cost: list[float] | None = None) -> tuple[bool, str]:
    with connect(DATABASE_PATH) as connection:

        # get the IDs of all papers that already exist
        paper = connection.execute(
            generate_sql_query('papers', columns=['paper_id'], conditions={'paper_id': paper_id})
        ).fetchone()

        # if the proposed paper does not exist, return an error message
        if not paper:
            return False, f"Paper {paper_id} does not exist. Please try adding it instead."

        # if a name is proposed, update the name of the paper
        if name is not None:
            connection.execute(
                "UPDATE papers SET name = ? WHERE paper_id = ?;",
                (name, paper_id)
            )
        
        # if delivery data is proposed, update the delivery data of the paper
        if days_delivered is not None:
            for day_id, delivered in enumerate(days_delivered):
                connection.execute(
                    "UPDATE papers_days_delivered SET delivered = ? WHERE paper_id = ? AND day_id = ?;",
                    (delivered, paper_id, day_id)
                )

        # if cost data is proposed, update the cost data of the paper
        if days_cost is not None:
            for day_id, cost in enumerate(days_cost):
                connection.execute(
                    "UPDATE papers_days_cost SET cost = ? WHERE paper_id = ? AND day_id = ?;",
                    (cost, paper_id, day_id)
                )

        connection.commit()

        return True, f"Paper {paper_id} edited."

    return False, "Something went wrong."


## delete an existing paper
 # do not allow if the paper does not exist
def delete_existing_paper(paper_id: int) -> tuple[bool, str]:
    with connect(DATABASE_PATH) as connection:

        # get the IDs of all papers that already exist
        paper = connection.execute(
            generate_sql_query('papers', columns=['paper_id'], conditions={'paper_id': paper_id})
        ).fetchone()

        # if the proposed paper does not exist, return an error message
        if not paper:
            return False, f"Paper {paper_id} does not exist. Please try adding it instead."

        # delete the paper from the names table
        connection.execute(
            "DELETE FROM papers WHERE paper_id = ?;",
            (paper_id, )
        )

        # delete the paper from the delivery data table
        connection.execute(
            "DELETE FROM papers_days_delivered WHERE paper_id = ?;",
            (paper_id, )
        )

        # delete the paper from the cost data table
        connection.execute(
            "DELETE FROM papers_days_cost WHERE paper_id = ?;",
            (paper_id, )
        )

        connection.commit()

        return True, f"Paper {paper_id} deleted."

    return False, "Something went wrong."

Add or delete undelivered strings

## record strings for date(s) paper(s) were not delivered
def add_undelivered_string(paper_id: int, undelivered_string: str, month: int, year: int) -> tuple[bool, str]:
    
    # if the string is not valid, return an error message
    if not validate_undelivered_string(undelivered_string):
        return False, f"Invalid undelivered string."
    
    with connect(DATABASE_PATH) as connection:
        # check if given paper exists
        paper = connection.execute(
            generate_sql_query(
                'papers',
                columns=['paper_id'],
                conditions={'paper_id': paper_id}
            )
        ).fetchone()

        # if the paper does not exist, return an error message
        if not paper:
            return False, f"Paper {paper_id} does not exist. Please try adding it instead."

        # check if a string with the same month and year, for the same paper, already exists
        existing_string = connection.execute(
            generate_sql_query(
                'undelivered_strings',
                columns=['string'],
                conditions={
                    'paper_id': paper_id,
                    'month': month,
                    'year': year
                }
            )
        ).fetchone()

        # if a string with the same month and year, for the same paper, already exists, concatenate the new string to it
        if existing_string:
            new_string = f"{existing_string[0]},{undelivered_string}"

            connection.execute(
                "UPDATE undelivered_strings SET string = ? WHERE paper_id = ? AND month = ? AND year = ?;",
                (new_string, paper_id, month, year)
            )

        # otherwise, add the new string to the database
        else:
            connection.execute(
                "INSERT INTO undelivered_strings (string, paper_id, month, year) VALUES (?, ?, ?, ?);",
                (undelivered_string, paper_id, month, year)
            )

        connection.commit()

    return True, f"Undelivered string added."


## delete an existing undelivered string
 # do not allow if the string does not exist
def delete_undelivered_string(paper_id: int, month: int, year: int) -> tuple[bool, str]:
    with connect(DATABASE_PATH) as connection:
    
        # check if a string with the same month and year, for the same paper, exists
        existing_string = connection.execute(
            generate_sql_query(
                'undelivered_strings',
                columns=['string'],
                conditions={
                    'paper_id': paper_id,
                    'month': month,
                    'year': year
                }
            )
        ).fetchone()

        # if it does, delete it
        if existing_string:
            connection.execute(
                "DELETE FROM undelivered_strings WHERE paper_id = ? AND month = ? AND year = ?;",
                (paper_id, month, year)
            )

            connection.commit()

            return True, f"Undelivered string deleted."

        # if the string does not exist, return an error message
        return False, f"Undelivered string does not exist."

    return False, "Something went wrong."

Extract data from user input

## extract delivery days and costs from user input
def extract_days_and_costs(days_delivered: str | None, prices: str | None, paper_id: int | None = None) -> tuple[list[bool], list[float]]:
    days = []
    costs = []

    # if the user has provided delivery days, extract them
    if days_delivered is not None:
        days = [
            bool(int(day == 'Y')) for day in str(days_delivered).upper()
        ]

    # if the user has not provided delivery days, fetch them from the database
    else:
        if isinstance(paper_id, int):
            days = [
                (int(day_id), bool(delivered))
                for day_id, delivered in query_database(
                    generate_sql_query(
                        'papers_days_delivered',
                        columns=['day_id', 'delivered'],
                        conditions={
                            'paper_id': paper_id
                        }
                    )
                )
            ]

            days.sort(key=lambda x: x[0])

            days = [delivered for _, delivered in days]

    # if the user has provided prices, extract them
    if prices is not None:

        costs = []
        encoded_prices = [float(price) for price in SPLIT_REGEX['semicolon'].split(prices.rstrip(';')) if float(price) > 0]

        day_count = -1
        for day in days:
            if day:
                day_count += 1
                cost = encoded_prices[day_count]

            else:
                cost = 0

            costs.append(cost)

    return days, costs

npbc_cli.py

Calculate the costs

## calculate the cost for a given month and year
 # default to the previous month if no month and no year is given
 # default to the current month if no month is given and year is given
 # default to the current year if no year is given and month is given
def calculate(args: arg_namespace) -> None:

    # deal with month and year
    if args.month or args.year:

        feedback = validate_month_and_year(args.month, args.year)

        if not feedback[0]:
            status_print(*feedback)
            return

        if args.month:
            month = args.month
        
        else:
            month = datetime.now().month

        if args.year:
            year = args.year

        else:
            year = datetime.now().year

    else:
        previous_month = get_previous_month()
        month = previous_month.month
        year = previous_month.year

    # look for undelivered strings in the database
    existing_strings = query_database(
        generate_sql_query(
            'undelivered_strings',
            columns=['paper_id', 'string'],
            conditions={
                'month': month,
                'year': year
            }
        )
    )

    # associate undelivered strings with their paper_id
    undelivered_strings: dict[int, str] = {
        paper_id: undelivered_string
        for paper_id, undelivered_string in existing_strings
    }

    # calculate the cost for each paper, as well as the total cost
    costs, total, undelivered_dates = calculate_cost_of_all_papers(
        undelivered_strings,
        month,
        year
    )

    # format the results
    formatted = format_output(costs, total, month, year)

    # unless the user specifies so, copy the results to the clipboard
    if not args.nocopy:
        copy_to_clipboard(formatted)

        formatted += '\nSummary copied to clipboard.'

    # unless the user specifies so, log the results to the database
    if not args.nolog:
        save_results(undelivered_dates, month, year)

        formatted += '\nLog saved to file.'

    # print the results
    status_print(True, "Success!")
    print(f"SUMMARY:\n{formatted}")

Get undelivered strings, paper data, or logs

## get undelivered strings from the database
 # filter by whichever parameter the user provides. they as many as they want.
 # available parameters: month, year, key, string
def getudl(args: arg_namespace) -> None:

    # validate the month and year
    feedback = validate_month_and_year(args.month, args.year)

    if not feedback[0]:
        status_print(*feedback)
        return
    
    conditions = {}

    if args.key:
        conditions['paper_id'] = args.key

    if args.month:
        conditions['month'] = args.month

    if args.year:
        conditions['year'] = args.year

    if args.undelivered:
        conditions['strings'] = str(args.undelivered).lower().strip()

        if not validate_undelivered_string(conditions['strings']):
            status_print(False, "Invalid undelivered string.")
            return

    # if the undelivered strings exist, fetch them
    undelivered_strings = query_database(
        generate_sql_query(
            'undelivered_strings',
            conditions=conditions
        )
    )

    # if there were undelivered strings, print them
    if undelivered_strings:
        status_print(True, 'Found undelivered strings.')

        print(f"{Fore.YELLOW}entry_id{Style.RESET_ALL} | {Fore.YELLOW}year{Style.RESET_ALL} | {Fore.YELLOW}month{Style.RESET_ALL} | {Fore.YELLOW}paper_id{Style.RESET_ALL} | {Fore.YELLOW}string{Style.RESET_ALL}")
        
        for string in undelivered_strings:
            print('|'.join([str(item) for item in string]))

    # otherwise, print that there were no undelivered strings
    else:
        status_print(False, 'No undelivered strings found.')

## get a list of all papers in the database
 # filter by whichever parameter the user provides. they may use as many as they want (but keys are always printed)
 # available parameters: name, days, costs
 # the output is provided as a formatted table, printed to the standard output
def getpapers(args: arg_namespace) -> None:
    headers = ['paper_id']

    # fetch a list of all papers' IDs
    papers_id_list = [
        paper_id
        for paper_id, in query_database(
            generate_sql_query(
                'papers',
                columns=['paper_id']
            )
        )
    ]

    # initialize lists for the data
    paper_name_list, paper_days_list, paper_costs_list = [], [], []

    # sort the papers' IDs (for the sake of consistency)
    papers_id_list.sort()

    # if the user wants names, fetch that data and add it to the list
    if args.names:

        # first get a dictionary of {paper_id: paper_name}
        papers_names = {
            paper_id: paper_name
            for paper_id, paper_name in query_database(
                generate_sql_query(
                    'papers',
                    columns=['paper_id', 'name']
                )
            )
        }

        # then use the sorted IDs list to create a sorted names list
        paper_name_list = [
            papers_names[paper_id]
            for paper_id in papers_id_list
        ]

        headers.append('name')

    # if the user wants delivery days, fetch that data and add it to the list
    if args.days:

        # initialize a dictionary of {paper_id: {day_id: delivery}}
        papers_days = {
            paper_id: {}
            for paper_id in papers_id_list
        }

        # then get the data for each paper
        for paper_id, day_id, delivered in query_database(
            generate_sql_query(
                'papers_days_delivered',
                columns=['paper_id', 'day_id', 'delivered']
            )
        ):
            papers_days[paper_id][day_id] = delivered

        # format the data so that it matches the regex pattern /^[YN]{7}$/, the same way the user must input this data
        paper_days_list = [
            ''.join([
                'Y' if int(papers_days[paper_id][day_id]) == 1 else 'N'
                for day_id, _ in enumerate(WEEKDAY_NAMES)
            ])
            for paper_id in papers_id_list
        ]

        headers.append('days')

    # if the user wants costs, fetch that data and add it to the list
    if args.prices:

        # initialize a dictionary of {paper_id: {day_id: price}}
        papers_costs = {
            paper_id: {}
            for paper_id in papers_id_list
        }

        # then get the data for each paper
        for paper_id, day_id, cost in query_database(
            generate_sql_query(
                'papers_days_cost',
                columns=['paper_id', 'day_id', 'cost']
            )
        ):
            papers_costs[paper_id][day_id] = cost

        # format the data so that it matches the regex pattern /^[x](;[x]){6}$/, where /x/ is a number that may be either a floating point or an integer, the same way the user must input this data.
        paper_costs_list = [
            ';'.join([
                str(papers_costs[paper_id][day_id])
                for day_id, _ in enumerate(WEEKDAY_NAMES)
            ])
            for paper_id in papers_id_list
        ]

        headers.append('costs')

    # print the headers
    print(' | '.join([
        f"{Fore.YELLOW}{header}{Style.RESET_ALL}"
        for header in headers
    ]))

    # print the data
    for index, paper_id in enumerate(papers_id_list):
        print(f"{paper_id}: ", end='')
        
        values = []

        if args.names:
            values.append(paper_name_list[index])

        if args.days:
            values.append(paper_days_list[index])

        if args.prices:
            values.append(paper_costs_list[index])

        print(', '.join(values))


## get a log of all deliveries for a paper
 # the user may specify parameters to filter the output by. they may use as many as they want, or none
 # available parameters: paper_id, month, year
def getlogs(args: arg_namespace) -> None:
    
    # validate the month and year
    feedback = validate_month_and_year(args.month, args.year)

    if not feedback[0]:
        status_print(*feedback)
        return
        
    conditions = {}

    # if the user specified a particular paper, add it to the conditions
    if args.key:
        conditions['paper_id'] = args.key

    if args.month:
        conditions['month'] = args.month

    if args.year:
        conditions['year'] = args.year

    # fetch the data
    undelivered_dates = query_database(
        generate_sql_query(
            'undelivered_dates',
            conditions=conditions
        )
    )

    # if data was found, print it
    if undelivered_dates:
        status_print(True, 'Success!')

        print(f"{Fore.YELLOW}entry_id{Style.RESET_ALL} | {Fore.YELLOW}year{Style.RESET_ALL} | {Fore.YELLOW}month{Style.RESET_ALL} | {Fore.YELLOW}paper_id{Style.RESET_ALL} | {Fore.YELLOW}dates{Style.RESET_ALL}")

        for date in undelivered_dates:
            print(', '.join(date))

    # if no data was found, print an error message
    else:
        status_print(False, 'No results found.')

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/6020a4f5db0bf40f54e35b725b305cfeafdd8f2b

\$\endgroup\$
1
  • 1
    \$\begingroup\$ @Reinderien Edited with the requested code. \$\endgroup\$ May 2, 2022 at 15:24

1 Answer 1

0
\$\begingroup\$

SQLite supports check constraints. You should use them to ensure that

  • delivered is only 0 or 1 (since booleans are not supported directly)
  • cost is greater than 0
  • year is between perhaps 2000 and 2100
  • month is between 0 and 11

Inline foreign key clauses are supported which are nicer to read than table-level clauses, so consider switching to them.

Compound primary keys are also supported. Isn't this:

CONSTRAINT unique_paper_day UNIQUE (paper_id, day_id)

really just your primary key?

day_id isn't particularly an ID. It's a day integer, similar to your year and month (so long as I understand your schema). A rename is in order, and this should also receive a check constraint.

You already use the connection as a context manager, which is good; but that implies a commit. That means that you should not need to commit manually.

I've said as much in your other question, but you shouldn't be dynamically generating queries as you are in generate_sql_query, and you need to delete return [].

This is problematic:

cost_and_delivery_data = [
    get_cost_and_delivery_data(paper_id)
    for paper_id, in papers # type: ignore
]

because get_cost_and_delivery_data will open and close the connection on every iteration. Instead, pass a connection in.

undelivered_dates.dates is a classic normalisation error. Newcomers to RDBMS often say "I want to store an array", and if they're determined enough (as you are), they might even succeed. But this is not a good idea. There should be a separate row in this table for each date.

get the names of all papers that already exist is a lie. That query doesn't fetch all names of papers that exist; it fetches the name of all existing papers whose name matches the given name. But more importantly, you don't actually care about the name; you only care about the existence of the row, so write

select exists(
    select 1 from papers where name=?
)

Further to that, your select exists and insert should not be separated, but should be in the same query. After executing it, check rowcount to see if anything was successfully inserted.

Even more combination: do not have a separate select for get the ID of the paper that was just added; that should also be in the same statement using a returning clause.

paper_days_cost and paper_days_delivered appear to have the same keys. If they have the same existence characteristics, they should be merged into one table.

I encourage you to work on the above, and (after some time has passed and you've gathered enough feedback from the community) post a new question with your updated code.

\$\endgroup\$
1
  • \$\begingroup\$ Thanks a lot! day_id (values 0-6) is an ID, for the day of the week. In my Core code, I've never defined "0 ==> Monday" or something like that, instead relying on day_name from the calendar module. The hope is that this won't tie the app down to my conventions, and adapt for the user (let Sunday be the first day of the week, for instance). day_id indexes list(day_name), functioning as an ID for that list. Agree with everything else. Not sure how I missed out normalizing, but I'll try to achieve 3NF before posting a revised version. \$\endgroup\$ May 3, 2022 at 10:24

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.