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: Newspaper Bill Calculator CLI with Python (1 of 3, Core)
- Post 2 of 3, CLI: Newspaper Bill Calculator CLI with Python (2 of 3, CLI)
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
- Each newspaper has a certain cost per day of the week
- Each newspaper may or may not be delivered on a given day
- Each newspaper has a name, and a number called a key
- You may register any dates when you didn't receive a paper in advance using the
addudlcommand - 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/--keyCLI 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 thegenerate_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
