1
\$\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.

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

npbc_core.py

"""
provides the core functionality
- sets up and communicates with the DB
- adds, deletes, edits, or retrieves data from the DB (such as undelivered strings, paper data, logs)
- performs the main calculations
- handles validation and parsing of many values (such as undelivered strings)
"""

from calendar import day_name as weekday_names_iterable
from calendar import monthcalendar, monthrange
from datetime import date as date_type
from datetime import datetime, timedelta
from os import environ
from pathlib import Path
from sqlite3 import Connection, connect
from typing import Generator

import npbc_exceptions
import npbc_regex

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


def get_number_of_each_weekday(month: int, year: int) -> Generator[int, None, None]:
    """generate a list of number of times each weekday occurs in a given month (return a generator)
    - the list will be in the same order as WEEKDAY_NAMES (so the first day should be Monday)"""

    # get the calendar for the month
    main_calendar = monthcalendar(year, month)

    # get the number of weeks in that month from the calendar
    number_of_weeks = len(main_calendar)

    # iterate over each possible weekday
    for i, _ in enumerate(WEEKDAY_NAMES):

        # assume that the weekday occurs once per week in the month
        number_of_weekday: int = number_of_weeks

        # if the first week doesn't have the weekday, decrement its count
        if main_calendar[0][i] == 0:
            number_of_weekday -= 1
        
        # if the last week doesn't have the weekday, decrement its count
        if main_calendar[-1][i] == 0:
            number_of_weekday -= 1

        yield number_of_weekday


def validate_undelivered_string(*strings: str) -> None:
    """validate a string that specifies when a given paper was not delivered
    - first check to see that it meets the comma-separated requirements
    - then check against each of the other acceptable patterns in the regex dictionary"""

    # check that the string matches one of the acceptable patterns
    for string in strings:
        if string and not (
            npbc_regex.NUMBER_MATCH_REGEX.match(string) or
            npbc_regex.RANGE_MATCH_REGEX.match(string) or
            npbc_regex.DAYS_MATCH_REGEX.match(string) or
            npbc_regex.N_DAY_MATCH_REGEX.match(string) or
            npbc_regex.ALL_MATCH_REGEX.match(string)
        ):

            raise npbc_exceptions.InvalidUndeliveredString(f'{string} is not a valid undelivered string.')

    # if we get here, all strings passed the regex check

def extract_number(string: str, month: int, year: int) -> date_type | None:
    """if the date is simply a number, it's a single day. so we just identify that date"""

    date = int(string)

    # if the date is valid for the given month
    if date > 0 and date <= monthrange(year, month)[1]:
        return date_type(year, month, date)


def extract_range(string: str, month: int, year: int) -> Generator[date_type, None, None]:
    """if the date is a range of numbers, it's a range of days. we identify all the dates in that range, bounds inclusive"""

    start, end = map(int, npbc_regex.HYPHEN_SPLIT_REGEX.split(string))

    # if the range is valid for the given month
    if 0 < start <= end <= monthrange(year, month)[1]:
        for date in range(start, end + 1):
            yield date_type(year, month, date)


def extract_weekday(string: str, month: int, year: int) -> Generator[date_type, None, None]:
    """if the date is the plural of a weekday name, we identify all dates in that month which are the given weekday"""

    weekday = WEEKDAY_NAMES.index(string.capitalize().rstrip('s'))

    for day in range(1, monthrange(year, month)[1] + 1):
        if date_type(year, month, day).weekday() == weekday:
            yield date_type(year, month, day)


def extract_nth_weekday(string: str, month: int, year: int) -> date_type | None:
    """if the date is a number and a weekday name (singular), we identify the date that is the nth occurrence of the given weekday in the month"""

    n, weekday_name = npbc_regex.HYPHEN_SPLIT_REGEX.split(string)

    n = int(n)

    # if the day is valid for the given month
    if n > 0 and n <= list(get_number_of_each_weekday(month, year))[WEEKDAY_NAMES.index(weekday_name.capitalize())]:
        
        # record the "day_id" corresponding to the given weekday name
        weekday = WEEKDAY_NAMES.index(weekday_name.capitalize())

        # store all dates when the given weekday occurs in the given month
        valid_dates = [
            date_type(year, month, day)
            for day in range(1, monthrange(year, month)[1] + 1)
            if date_type(year, month, day).weekday() == weekday
        ]

        # return the date that is the nth occurrence of the given weekday in the month
        return valid_dates[n - 1]


def extract_all(month: int, year: int) -> Generator[date_type, None, None]:
    """if the text is "all", we identify all the dates in the month"""

    for day in range(1, monthrange(year, month)[1] + 1):
        yield date_type(year, month, day)


def parse_undelivered_string(month: int, year: int, string: str) -> set[date_type]:
    """parse a section of the strings
    - each section is a string that specifies a set of dates
    - this function will return a set of dates that uniquely identifies each date mentioned across the string"""

    # initialize the set of dates
    dates = set()

    # check for each of the patterns
    if npbc_regex.NUMBER_MATCH_REGEX.match(string):
        number_date = extract_number(string, month, year)

        if number_date:
            dates.add(number_date)

    elif npbc_regex.RANGE_MATCH_REGEX.match(string):
        dates.update(extract_range(string, month, year))

    elif npbc_regex.DAYS_MATCH_REGEX.match(string):
        dates.update(extract_weekday(string, month, year))

    elif npbc_regex.N_DAY_MATCH_REGEX.match(string):
        n_day_date = extract_nth_weekday(string, month, year)

        if n_day_date:
            dates.add(n_day_date)

    elif npbc_regex.ALL_MATCH_REGEX.match(string):
        dates.update(extract_all(month, year))

    else:
        raise npbc_exceptions.InvalidUndeliveredString(f'{string} is not a valid undelivered string.')

    return dates

    
def parse_undelivered_strings(month: int, year: int, *strings: str) -> set[date_type]:
    """parse a string that specifies when a given paper was not delivered
    - each section states some set of dates
    - this function will return a set of dates that uniquely identifies each date mentioned across all the strings"""
    
    # initialize the set of dates
    dates = set()

    # check for each of the patterns
    for string in strings:
        try:
            dates.update(parse_undelivered_string(month, year, string))

        except npbc_exceptions.InvalidUndeliveredString:
            print(
                f"""Congratulations! You broke the program!
                You managed to write a string that the program considers valid, but isn't actually.
                Please report it to the developer.
                \nThe string you wrote was: {string}
                This data has not been counted."""
            )

    return dates


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


def calculate_cost_of_one_paper(
        number_of_each_weekday: list[int],
        undelivered_dates: set[date_type],
        cost_and_delivered_data: list[tuple[bool, float]]
    ) -> float:
    """calculate the cost of one paper for the full month
    - any dates when it was not delivered will be removed"""
    
    # initialize counters corresponding to each weekday when the paper was not delivered
    number_of_days_per_weekday_not_received = [0] * len(number_of_each_weekday)
    
    # for each date that the paper was not delivered, we increment the counter for the corresponding weekday
    for date in undelivered_dates:
        number_of_days_per_weekday_not_received[date.weekday()] += 1

    # calculate the total number of each weekday the paper was delivered (if it is supposed to be delivered)
    # multiply this number by the cost of each day
    # add all the costs together and return the result
    return sum([
        (number_of_each_weekday[day_id] - number_of_days_per_weekday_not_received[day_id]) * cost
        if delivered else 0
        for day_id, (delivered, cost) in enumerate(cost_and_delivered_data)
    ])


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


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


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


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


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



def get_previous_month() -> date_type:
    """get the previous month, by looking at 1 day before the first day of the current month (duh)"""

    return (datetime.today().replace(day=1) - timedelta(days=1)).replace(day=1)


def validate_month_and_year(month: int | None = None, year: int | None = None) -> None:
    """validate month and year
    - month must be an integer between 1 and 12 inclusive
    - year must be an integer greater than 0"""
    
    if isinstance(month, int) and not (1 <= month <= 12):
        raise npbc_exceptions.InvalidMonthYear("Month must be between 1 and 12.")

    if isinstance(year, int) and (year <= 0):
        raise npbc_exceptions.InvalidMonthYear("Year must be greater than 0.")

test_core.py

"""
test data-independent functions from the core
- none of these depend on data in the database
"""

from datetime import date as date_type

from pytest import raises

import npbc_core
from npbc_exceptions import InvalidMonthYear, InvalidUndeliveredString


def test_get_number_of_each_weekday():
    test_function = npbc_core.get_number_of_each_weekday

    assert list(test_function(1, 2022)) == [5, 4, 4, 4, 4, 5, 5]
    assert list(test_function(2, 2022)) == [4, 4, 4, 4, 4, 4, 4]
    assert list(test_function(3, 2022)) == [4, 5, 5 ,5, 4, 4, 4]
    assert list(test_function(2, 2020)) == [4, 4, 4, 4, 4, 5, 4]
    assert list(test_function(12, 1954)) == [4, 4, 5, 5, 5, 4, 4]


def test_validate_undelivered_string():
    test_function = npbc_core.validate_undelivered_string

    with raises(InvalidUndeliveredString):
        test_function("a")
        test_function("monday")
        test_function("1-mondays")
        test_function("1monday")
        test_function("1 monday")
        test_function("monday-1")
        test_function("monday-1")

    test_function("")
    test_function("1")
    test_function("6")
    test_function("31")
    test_function("31","")
    test_function("3","1")
    test_function("3","1","")
    test_function("3","1")
    test_function("3","1")
    test_function("3","1")
    test_function("1","2","3-9")
    test_function("1","2","3-9","11","12","13-19")
    test_function("1","2","3-9","11","12","13-19","21","22","23-29")
    test_function("1","2","3-9","11","12","13-19","21","22","23-29","31")
    test_function("1","2","3","4","5","6","7","8","9")
    test_function("mondays")
    test_function("mondays,tuesdays")
    test_function("mondays","tuesdays","wednesdays")
    test_function("mondays","5-21")
    test_function("mondays","5-21","tuesdays","5-21")
    test_function("1-monday")
    test_function("2-monday")
    test_function("all")
    test_function("All")
    test_function("aLl")
    test_function("alL")
    test_function("aLL")
    test_function("ALL")


def test_undelivered_string_parsing():
    MONTH = 5
    YEAR = 2017
    test_function = npbc_core.parse_undelivered_strings


    assert test_function(MONTH, YEAR, '') == set([])

    assert test_function(MONTH, YEAR, '1') == set([
        date_type(year=YEAR, month=MONTH, day=1)
    ])

    assert test_function(MONTH, YEAR, '1-2') == set([
        date_type(year=YEAR, month=MONTH, day=1),
        date_type(year=YEAR, month=MONTH, day=2)
    ])

    assert test_function(MONTH, YEAR, '5-17') == set([
        date_type(year=YEAR, month=MONTH, day=5),
        date_type(year=YEAR, month=MONTH, day=6),
        date_type(year=YEAR, month=MONTH, day=7),
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=9),
        date_type(year=YEAR, month=MONTH, day=10),
        date_type(year=YEAR, month=MONTH, day=11),
        date_type(year=YEAR, month=MONTH, day=12),
        date_type(year=YEAR, month=MONTH, day=13),
        date_type(year=YEAR, month=MONTH, day=14),
        date_type(year=YEAR, month=MONTH, day=15),
        date_type(year=YEAR, month=MONTH, day=16),
        date_type(year=YEAR, month=MONTH, day=17)
    ])

    assert test_function(MONTH, YEAR, '5-17', '19') == set([
        date_type(year=YEAR, month=MONTH, day=5),
        date_type(year=YEAR, month=MONTH, day=6),
        date_type(year=YEAR, month=MONTH, day=7),
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=9),
        date_type(year=YEAR, month=MONTH, day=10),
        date_type(year=YEAR, month=MONTH, day=11),
        date_type(year=YEAR, month=MONTH, day=12),
        date_type(year=YEAR, month=MONTH, day=13),
        date_type(year=YEAR, month=MONTH, day=14),
        date_type(year=YEAR, month=MONTH, day=15),
        date_type(year=YEAR, month=MONTH, day=16),
        date_type(year=YEAR, month=MONTH, day=17),
        date_type(year=YEAR, month=MONTH, day=19)
    ])

    assert test_function(MONTH, YEAR, '5-17', '19-21') == set([
        date_type(year=YEAR, month=MONTH, day=5),
        date_type(year=YEAR, month=MONTH, day=6),
        date_type(year=YEAR, month=MONTH, day=7),
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=9),
        date_type(year=YEAR, month=MONTH, day=10),
        date_type(year=YEAR, month=MONTH, day=11),
        date_type(year=YEAR, month=MONTH, day=12),
        date_type(year=YEAR, month=MONTH, day=13),
        date_type(year=YEAR, month=MONTH, day=14),
        date_type(year=YEAR, month=MONTH, day=15),
        date_type(year=YEAR, month=MONTH, day=16),
        date_type(year=YEAR, month=MONTH, day=17),
        date_type(year=YEAR, month=MONTH, day=19),
        date_type(year=YEAR, month=MONTH, day=20),
        date_type(year=YEAR, month=MONTH, day=21)
    ])

    assert test_function(MONTH, YEAR, '5-17', '19-21', '23') == set([
        date_type(year=YEAR, month=MONTH, day=5),
        date_type(year=YEAR, month=MONTH, day=6),
        date_type(year=YEAR, month=MONTH, day=7),
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=9),
        date_type(year=YEAR, month=MONTH, day=10),
        date_type(year=YEAR, month=MONTH, day=11),
        date_type(year=YEAR, month=MONTH, day=12),
        date_type(year=YEAR, month=MONTH, day=13),
        date_type(year=YEAR, month=MONTH, day=14),
        date_type(year=YEAR, month=MONTH, day=15),
        date_type(year=YEAR, month=MONTH, day=16),
        date_type(year=YEAR, month=MONTH, day=17),
        date_type(year=YEAR, month=MONTH, day=19),
        date_type(year=YEAR, month=MONTH, day=20),
        date_type(year=YEAR, month=MONTH, day=21),
        date_type(year=YEAR, month=MONTH, day=23)
    ])

    assert test_function(MONTH, YEAR, 'mondays') == set([
        date_type(year=YEAR, month=MONTH, day=1),
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=15),
        date_type(year=YEAR, month=MONTH, day=22),
        date_type(year=YEAR, month=MONTH, day=29)
    ])

    assert test_function(MONTH, YEAR, 'mondays', 'wednesdays') == set([
        date_type(year=YEAR, month=MONTH, day=1),
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=15),
        date_type(year=YEAR, month=MONTH, day=22),
        date_type(year=YEAR, month=MONTH, day=29),
        date_type(year=YEAR, month=MONTH, day=3),
        date_type(year=YEAR, month=MONTH, day=10),
        date_type(year=YEAR, month=MONTH, day=17),
        date_type(year=YEAR, month=MONTH, day=24),
        date_type(year=YEAR, month=MONTH, day=31)
    ])

    assert test_function(MONTH, YEAR, '2-monday') == set([
        date_type(year=YEAR, month=MONTH, day=8)
    ])

    assert test_function(MONTH, YEAR, '2-monday', '3-wednesday') == set([
        date_type(year=YEAR, month=MONTH, day=8),
        date_type(year=YEAR, month=MONTH, day=17)
    ])


def test_calculating_cost_of_one_paper():
    DAYS_PER_WEEK = [5, 4, 4, 4, 4, 5, 5]

    COST_AND_DELIVERY_DATA: list[tuple[bool, float]] = [
        (False, 0),
        (False, 0),
        (True, 2),
        (True, 2),
        (True, 5),
        (False, 0),
        (True, 1)
    ]

    test_function = npbc_core.calculate_cost_of_one_paper

    assert test_function(
        DAYS_PER_WEEK,
        set([]),
        COST_AND_DELIVERY_DATA
    ) == 41

    assert test_function(
        DAYS_PER_WEEK,
        set([]),
        [
            (False, 0),
            (False, 0),
            (True, 2),
            (True, 2),
            (True, 5),
            (False, 0),
            (False, 1)
        ]
    ) == 36

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=8)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 41

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=8),
            date_type(year=2022, month=1, day=8)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 41

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=8),
            date_type(year=2022, month=1, day=17)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 41

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=2)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 40

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=2),
            date_type(year=2022, month=1, day=2)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 40

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=6),
            date_type(year=2022, month=1, day=7)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 34

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=6),
            date_type(year=2022, month=1, day=7),
            date_type(year=2022, month=1, day=8)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 34

    assert test_function(
        DAYS_PER_WEEK,
        set([
            date_type(year=2022, month=1, day=6),
            date_type(year=2022, month=1, day=7),
            date_type(year=2022, month=1, day=7),
            date_type(year=2022, month=1, day=7),
            date_type(year=2022, month=1, day=8),
            date_type(year=2022, month=1, day=8),
            date_type(year=2022, month=1, day=8)
        ]),
        COST_AND_DELIVERY_DATA
    ) == 34


def test_validate_month_and_year():
    test_function = npbc_core.validate_month_and_year

    test_function(1, 2020)
    test_function(12, 2020)
    test_function(1, 2021)
    test_function(12, 2021)
    test_function(1, 2022)
    test_function(12, 2022)

    with raises(InvalidMonthYear):
        test_function(-54, 2020)
        test_function(0, 2020)
        test_function(13, 2020)
        test_function(45, 2020)
        test_function(1, -5)
        test_function(12, -5)
        test_function(1.6, 10) # type: ignore
        test_function(12.6, 10) # type: ignore
        test_function(1, '10') # type: ignore
        test_function(12, '10') # type: ignore

npbc_regex.py

"""
regex used by other files
- MATCH regex are used to validate (usually user input)
- SPLIT regex are used to split strings (usually user input)
"""

from calendar import day_name as WEEKDAY_NAMES_ITERABLE
from re import compile as compile_regex


## regex used to match against strings

# match for a list of comma separated values. each value must be/contain digits, or letters, or hyphens. spaces are allowed between values and commas. any number of values are allowed, but at least one must be present.
CSV_MATCH_REGEX = compile_regex(r'^[-\w]+( *, *[-\w]+)*( *,)?$')

# match for a single number. must be one or two digits
NUMBER_MATCH_REGEX = compile_regex(r'^[\d]{1,2}?$')

# match for a range of numbers. each number must be one or two digits. numbers are separated by a hyphen. spaces are allowed between numbers and the hyphen.
RANGE_MATCH_REGEX = compile_regex(r'^\d{1,2} *- *\d{1,2}$')

# match for weekday name. day must appear as "daynames" (example = "mondays"). all lowercase.
DAYS_MATCH_REGEX = compile_regex(f"^{'|'.join(map(lambda x: x.lower() + 's', WEEKDAY_NAMES_ITERABLE))}$")

# match for nth weekday name. day must appear as "n-dayname" (example = "1-monday"). all lowercase. must be one digit.
N_DAY_MATCH_REGEX = compile_regex(f"^\\d *- *({'|'.join(map(lambda x: x.lower(), WEEKDAY_NAMES_ITERABLE))})$")

# match for the text "all" in any case.
ALL_MATCH_REGEX = compile_regex(r'^[aA][lL]{2}$')

# match for seven values, each of which must be a 'Y' or an 'N'. there are no delimiters.
DELIVERY_MATCH_REGEX = compile_regex(r'^[YN]{7}$')


## regex used to split strings

# split on hyphens. spaces are allowed between hyphens and values.
HYPHEN_SPLIT_REGEX = compile_regex(r' *- *')

test_regex.py

import npbc_regex

def test_regex_number():
    assert npbc_regex.NUMBER_MATCH_REGEX.match('') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('1') is not None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('1 2') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('1-2') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('11') is not None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('11-12') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('11-12,13') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('11-12,13-14') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('111') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('a') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('1a') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('1a2') is None
    assert npbc_regex.NUMBER_MATCH_REGEX.match('12b') is None

def test_regex_range():
    assert npbc_regex.RANGE_MATCH_REGEX.match('') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('1') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('1 2') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('1-2') is not None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-12') is not None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-12-1') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11 -12') is not None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11 - 12') is not None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11- 12') is not None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-2') is not None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-12,13') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-12,13-14') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('111') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('a') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('1a') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('1a2') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('12b') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-a') is None
    assert npbc_regex.RANGE_MATCH_REGEX.match('11-12a') is None

def test_regex_CSVs():
    assert npbc_regex.CSV_MATCH_REGEX.match('') is None
    assert npbc_regex.CSV_MATCH_REGEX.match('1') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('a') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('adcef') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('-') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match(' ') is None
    assert npbc_regex.CSV_MATCH_REGEX.match('1,2') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('1-3') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('monday') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('monday,tuesday') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('mondays') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('tuesdays') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('1,2,3') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('1-3') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('monday,tuesday') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match('mondays,tuesdays') is not None
    assert npbc_regex.CSV_MATCH_REGEX.match(';') is None
    assert npbc_regex.CSV_MATCH_REGEX.match(':') is None
    assert npbc_regex.CSV_MATCH_REGEX.match(':') is None
    assert npbc_regex.CSV_MATCH_REGEX.match('!') is None
    assert npbc_regex.CSV_MATCH_REGEX.match('1,2,3,4') is not None

def test_regex_days():
    assert npbc_regex.DAYS_MATCH_REGEX.match('') is None
    assert npbc_regex.DAYS_MATCH_REGEX.match('1') is None
    assert npbc_regex.DAYS_MATCH_REGEX.match('1,2') is None
    assert npbc_regex.DAYS_MATCH_REGEX.match('1-3') is None
    assert npbc_regex.DAYS_MATCH_REGEX.match('monday') is None
    assert npbc_regex.DAYS_MATCH_REGEX.match('monday,tuesday') is None
    assert npbc_regex.DAYS_MATCH_REGEX.match('mondays') is not None
    assert npbc_regex.DAYS_MATCH_REGEX.match('tuesdays') is not None

def test_regex_n_days():
    assert npbc_regex.N_DAY_MATCH_REGEX.match('') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1-') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1,2') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1-3') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('monday') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('monday,tuesday') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('mondays') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1-tuesday') is not None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('11-tuesday') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('111-tuesday') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('11-tuesdays') is None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1 -tuesday') is not None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1- tuesday') is not None
    assert npbc_regex.N_DAY_MATCH_REGEX.match('1 - tuesday') is not None

def test_regex_all_text():
    assert npbc_regex.ALL_MATCH_REGEX.match('') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1-') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1,2') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1-3') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('monday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('monday,tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('mondays') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('tuesdays') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1-tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('11-tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('111-tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('11-tuesdays') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1 -tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1- tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('1 - tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('all') is not None
    assert npbc_regex.ALL_MATCH_REGEX.match('all,tuesday') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('all,tuesdays') is None
    assert npbc_regex.ALL_MATCH_REGEX.match('All') is not None
    assert npbc_regex.ALL_MATCH_REGEX.match('AlL') is not None
    assert npbc_regex.ALL_MATCH_REGEX.match('ALL') is not None

def test_delivery_regex():
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('a') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1.') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1.5') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1,2') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1-2') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1;2') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1:2') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('1,2,3') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('Y') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('N') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYYY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYYYY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYYYYY') is not None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYYYYYY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('NNNNNNN') is not None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('NYNNNNN') is not None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('NYYYYNN') is not None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('NYYYYYY') is not None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('NYYYYYYY') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('N,N,N,N,N,N,N') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('N;N;N;N;N;N;N') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('N-N-N-N-N-N-N') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('N N N N N N N') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYYYYy') is None
    assert npbc_regex.DELIVERY_MATCH_REGEX.match('YYYYYYn') is None



def test_regex_hyphen():
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1-2') == ['1', '2']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1-2-3') == ['1', '2', '3']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1 -2-3') == ['1', '2', '3']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1 - 2-3') == ['1', '2', '3']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1- 2-3') == ['1', '2', '3']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1') == ['1']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1-') == ['1', '']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1-2-') == ['1', '2', '']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1-2-3-') == ['1', '2', '3', '']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1,2-3') == ['1,2', '3']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1,2-3-') == ['1,2', '3', '']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('1,2, 3,') == ['1,2, 3,']
    assert npbc_regex.HYPHEN_SPLIT_REGEX.split('') == ['']

npbc_exceptions.py

"""
provide exceptions for other modules
- these are custom exceptions used to make error handling easier
- none of them inherit from BaseException
"""


from sqlite3 import OperationalError

class InvalidInput(ValueError): ...
class InvalidUndeliveredString(InvalidInput): ...
class PaperAlreadyExists(OperationalError): ...
class PaperNotExists(OperationalError): ...
class StringNotExists(OperationalError): ...
class InvalidMonthYear(InvalidInput): ...
class NoParameters(ValueError): ...

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

\$\endgroup\$

2 Answers 2

0
\$\begingroup\$

Pretty good!

This pattern:

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

is a little risky. Rather than having secondary flags (DEVELOPMENT) that derive other settings (DATABASE_DIR) in the code, have the code fetch those settings (DATABASE_DIR) directly from the environment. The development environment will set those settings differently from the production environment. You ask:

I don't understand what you mean when you say "risky" about environment variables—do you mean that I might get an error if they're not defined? I wanted to leverage the CI variable built into both GitHub and GitLab CI/CD.

Leveraging that variable is not a good idea. Applications should be categorically unable to vary their behaviour based on knowledge of which environment they're in; they should only pay attention to well-defined configuration points like your database directory. There will soon come a time when you have a problem, and you need to understand why the problem manifests one way in production and a different way in development. Configuration between these environments will naturally vary, but the application must be unable to vary its behaviour based on direct knowledge of which environment it's in or else there will be a complexity explosion that makes such debugging more difficult. It's easy to use GitHub actions to expose a different set of configuration variables based on which environment you're in.

I'd expect, rather than .db, to see .sqlite as the database extension; though this is minor.

For WEEKDAY_NAMES use a tuple instead of a list as the former is immutable.

Put your connection.close in a finally.

for i, _ in enumerate is not a good use of enumerate; instead just use range(len.

This:

    if string and not (
        npbc_regex.NUMBER_MATCH_REGEX.match(string) or
        npbc_regex.RANGE_MATCH_REGEX.match(string) or
        npbc_regex.DAYS_MATCH_REGEX.match(string) or
        npbc_regex.N_DAY_MATCH_REGEX.match(string) or
        npbc_regex.ALL_MATCH_REGEX.match(string)
    ):

can be expressed as

if string and not any(
    pattern.match(string) for pattern in (
        npbc_regex.NUMBER_MATCH_REGEX,
        npbc_regex.RANGE_MATCH_REGEX,
        npbc_regex.DAYS_MATCH_REGEX,
        npbc_regex.N_DAY_MATCH_REGEX,
        npbc_regex.ALL_MATCH_REGEX,
    )
):

It's odd to alias date as date_type. Probably don't do this.

if date > 0 and date <= monthrange(year, month)[1]: can use the if 0 < date <= monthrange ... syntax.

I find list(map to be less legible than a plain old comprehension, as in

[
    (bool(row[0]), row[1])
    for row in connection.execute(query, (paper_id,)).fetchall()
]

This:

(custom_timestamp if custom_timestamp else datetime.now())

can just be

(custom_timestamp or datetime.now())

Broadly, you have an issue with connection management. You open and close your connection too often. You should open it only once per program run. You ask:

Can you advise how to manage connections? Should I make a global variable or try to pass around function arguments? I would have done it in main() but Core does not have a main()

Identify the highest point in the application stack where a connection is needed. Spin it up there, don't put it in globals, and pass it to the functions that need it.

Use tuples instead of lists in your test functions.

\$\endgroup\$
2
  • 1
    \$\begingroup\$ Thank you. Doubts: 1 I don't understand what you mean when you say "risky" about environment variables—do you mean that I might get an error if they're not defined? I wanted to leverage the CI variable built into both GitHub and GitLab CI/CD. 2 DB connection is already in a context, but context doesn't call close() according to docs. finally is a good idea. 3 Can you advise how to manage connections? Should I make a global variable or try to pass around function arguments? I would have done it in main() but Core does not have a main() 4 any is really cool, thanks! \$\endgroup\$ May 30 at 14:07
  • 1
    \$\begingroup\$ @eccentricOrange edited. \$\endgroup\$
    – Reinderien
    May 30 at 15:17
1
\$\begingroup\$

typing.Generator is Deprecated

As of Python 3.9, from typing import Generator is deprecated. Instead, import it from collections.abc. (See PEP 585).

Unicode

Your regexes seem to assume strings of ASCII characters and they are only tested with characters in the ASCII range. However, in Python 3.x strings are Unicode and the re module operates on Unicode strings by default. For example, \d matches unicode digits and npbc_regex.NUMBER_MATCH_REGEX.match('੩੬') would return a match object. The comments suggest the strings come from user input, so stricter validation and testing may be warranted.

Regex

Regex patterns can be hard to decipher. This pattern:

compile_regex(r'^[aA][lL]{2}$')

probably takes longer to mentally parse than this one:

compile_regex(r'^[aA][lL][lL]$').

Or better yet, this clearly expresses the intent of the regex:

compile_regex(r'^all$', re.IGNORECASE) 

Generators vs Lists.

Every call to get_number_of_each_weekday() is inside a call to list(). It might make more sense to just return a list instead.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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