SQLite 3 database functions

I have the following function that I have placed in a db.py file:

def create_habit(self, name: str, period: str):
        self.cursor.execute(
            'INSERT INTO habits (name, period, created_at) VALUES (?, ?, ?)',
            (name, period, datetime.now())
        )
        self.conn.commit()

I would like this function to collect a list of habits. Additionally, I have a separate file that runs an interface where users can create habits using the above-mentioned function that looks like this:

@cli.command()
def create():
    name = questionary.text("Enter the habit name: ").ask()
    period = questionary.select("Enter the habit period:", choices=["daily", "weekly", "monthly"]).ask()
    tracker.create_habit(name, period)
    click.echo(f'Habit "{name}" with period "{period}" created successfully!')

The problem that I am having is that when I run the interface, the create_habit function returns an empty list. There is no traceback or error message, but the list is completely empty. Where am I going wrong? Any assistance would be greatly appreciated!

I do not see a list in the code extracts.

What exactly do you mean?
Nothing is added to the sqlite database?
Something else?

What I mean is that the create_habit function is supposed to add an entry to the database that has a name, period, and creation date. How’s,.that is not being added to the database. When I go to return that information, then the database doesn’t return an entry.

*however that is not being added to the database

Write a short python script that does the minimum to reproduce this problem.
Should be only a few lines of code.
Does that work? If so figure out why the small version is different to the main program.
If it fails and you are sure why share that code for us to run.

What version of python are you using and on which OS?

I did something similar. I wrote a brief tester program, and it all seemed okay. I actually removed the logic from the classes I wrote and the functions, and it worked without issue. The problem is that I am trying to do this using OOP in an interactive app. I am using Python 3.8 on a Linux Mint 20 machine.

Without an error or some other clue i am not sure how to help.

Thanks for your help. I really appreciate everything. This is my app in question:

I was just wondering if exceptions are being suppressed in the create_habit function.
Try adding prints after every statement and see if they appear.

I see the success message, but I do not see any call to the establish_a_connection function that sets up the DB.

Then I looked a bit deeper and I cannot find where you use the HabitDB class.
All I can see is this:

$ grep HabitDB *.py
db.py:class HabitDB:
main.py:from db import HabitDB

Therefore you cannot expect there to be a .db file to exists and have any content.

Thanks for the assistance. I thought that when I imported my HabitDB class into my main.py file that it would use those methods contained in the class. This is my first OOP project. Other than importing that class into my main.py file, how can I incorporate my HabitDB class?

You have to have make an instance of the class first.

db = HabitDB()

Then you have to call the methods on the instance.

db.establish_a_connection()

I would also recommend that you add an __init__ to HabitDB and use it to set the variables used in the class to None. This will help you catch errors.

1 Like

Not that my intention is to gate crash this thread, but more as an example of and build on what @barry-scott has posted; without which I would not have thought of this:

class user_id:
    def __init__(self, name, uid):
        self.name = name
        self.uid = uid
    def get_user(uid):
        print ('User '+uid.uid+' is '+uid.name)

Maybe this helps; maybe not.

Thanks a lot for your assistance guys. I really appreciate your patience with this noob!

Went back and rewrote some of my files. This is my new main.py file:

import click
from habit import HabitTracker, Habit
import questionary
from analytics import *
from db2 import HabitDB, HabitTracker

tracker = HabitTracker()



@click.group()
def cli():
    pass

@cli.command()
def create():
    name = questionary.text("Enter the habit name: ").ask()
    period = questionary.select("Enter the habit period:", choices=["daily", "weekly", "monthly"]).ask()
    tracker.create_habit(name, period)
    click.echo(f'Habit "{name}" with period "{period}" created successfully!')

@cli.command()
def delete():
    name = questionary.text("Enter the habit name: ").ask()
    tracker.delete_habit(name)
    click.echo(f'Habit "{name}" deleted successfully!')

@cli.command()
def list_habit_groups():
    habits = tracker.get_habits()
    click.echo('Current habits:')
    for habit in habits:
        click.echo(f'- {habit.name} ({habit.period})')

@cli.command()
def list_habit_groups_period():
    period = questionary.select("Enter the habit period:", choices=["daily", "weekly", "monthly"]).ask()
    habits = tracker.get_habits_by_period(period)
    click.echo(f'Current {period} habits:')
    for habit in habits:
        click.echo(f'- {habit.name}')

@cli.command()
def longest_streak():
    longest_streak_habit = tracker.get_longest_streak()
    if longest_streak_habit:
        click.echo(f'Habit with longest streak: {longest_streak_habit.name} ({longest_streak_habit.get_streak()})')
    else:
        click.echo('No habits with a streak.')

@cli.command()
def longest_streak_habit():
    name = questionary.text("Enter the habit name: ").ask()
    streak = tracker.get_longest_streak_by_habit(name)
    if streak:
        click.echo(f'Longest streak for habit "{name}": {streak}')
    else:
        click.echo(f'Habit "{name}" not found.')

@cli.command()
def mark():
    name = questionary.text("Enter the habit name: ").ask()
    tracker.mark_complete(name)
    click.echo(f'Habit "{name}" marked successfully!')

@cli.command()
def unmark():
    name = questionary.text("Enter the habit name: ").ask()
    tracker.mark_incomplete(name)
    click.echo(f'Habit "{name}" unmarked successfully!')

def main():
    while True:
        command = input('Enter a command (create, delete, list, list-period, longest-streak, longest-streak-habit, or exit): ')
        if command == 'create':
            name = input('Enter the habit name: ')
            period = input('Enter the habit period (daily or weekly): ')
            tracker.create_habit(name, period)
            click.echo(f'Habit "{name}" with period "{period}" created successfully!')
        elif command == 'delete':
            name = input('Enter the habit name: ')
            tracker.delete_habit(name)
            click.echo(f'Habit "{name}" deleted successfully!')
        elif command == 'habit_groups':
            habits = tracker.get_habits()
            click.echo('Current habits:')
            for habit in habits:
                click.echo(f'- {habit.name} ({habit.period})')
        elif command == 'habit_groups_period':
            period = input('Enter the habit period (daily or weekly): ')
            habits = tracker.get_habits_by_period(period)
            click.echo(f'Current {period} habits:')
            for habit in habits:
                click.echo(f'- {habit.name}')
        elif command == 'longest-streak':
            longest_streak_habit = tracker.get_longest_streak()
            if longest_streak_habit:
                click.echo(f'Habit with longest streak: {longest_streak_habit.name} ({longest_streak_habit.get_streak()})')
            else:
                click.echo('No habits with a streak.')
        elif command == 'longest-streak-habit':
            name = input('Enter the habit name: ')
            streak = tracker.get_longest_streak_by_habit(name)
            if streak:
                click.echo(f'Longest streak for habit "{name}": {streak}')
            else:
                click.echo(f'Habit "{name}" not found.')
        elif command == 'mark':
            name = input('Enter the habit name: ')
            tracker.mark_complete(name)
            click.echo(f'Habit "{name}" marked successfully!')
        elif command == 'unmark':
            name = input('Enter the habit name: ')
            tracker.mark_incomplete(name)
            click.echo(f'Habit "{name}" unmarked successfully!')
        elif command == 'exit':
            break

if __name__ == '__main__':
    cli()
    main()

My new db.py file:

import sqlite3
import datetime
from habit import Habit, HabitTracker


class HabitDB:
    def establish_a_connection(self):
        self.conn = sqlite3.connect('habits.db')
        self.cursor = self.conn.cursor()

    # Create habits table if it does not exist
        self.cursor.execute('''
                CREATE TABLE IF NOT EXISTS habits (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                period TEXT NOT NULL,
                created_at DATETIME NOT NULL
            )''')

    # Create completions table if it does not exist 
        self.cursor.execute(''' 
                CREATE TABLE IF NOT EXISTS completions ( 
                habit_id INTEGER NOT NULL, 
                completed_at DATETIME NOT NULL, 
                FOREIGN KEY(habit_id) REFERENCES habits(id) ON 
                DELETE CASCADE  //added ON DELETE CASCADE to delete the related entries in completions when a row in habits is deleted 
            )''')
        self.conn.commit()

    def create_habit(self, name: str, period: str):
        self.cursor.execute(
            'INSERT INTO habits (name, period, created_at) VALUES (?, ?, ?)',
            (name, period, datetime.now())
        )
        self.conn.commit()

    def delete_habit(self, name: str):
        self.cursor.execute(
            'DELETE FROM habits WHERE name=?',
            (name,)
        )
        self.conn.commit()

    def mark_complete(self, name: str):
        self.cursor.execute(
            'SELECT id FROM habits WHERE name=?',
            (name,)
        )
        habit_id = self.cursor.fetchone()[0]
        self.cursor.execute(
            'INSERT INTO completions (habit_id, completed_at) VALUES (?, ?)',
            (habit_id, datetime.now())
        )
        self.conn.commit()

    def mark_incomplete(self, name: str):
        self.cursor.execute(
            'SELECT id FROM habits WHERE name=?',
            (name,)
        )
        habit_id = self.cursor.fetchone()[0]
        self.cursor.execute(
            'DELETE FROM completions WHERE habit_id=? ORDER BY completed_at DESC LIMIT 1',
            (habit_id,)
        )
        self.conn.commit()

    
    def get_habits(self):
        self.cursor.execute('SELECT * FROM habits')
        rows = self.cursor.fetchall()
        habits = []
        for row in rows:
            id, name, period, createdat = row
            self.cursor.execute(
            'SELECT completedat FROM completions WHERE habitid=?',
                (id,))
            completedatrows = self.cursor.fetchall()
            completedat = [row[0] for row in completedatrows]
            habits.append(Habit(id, name, period, createdat, completedat))
        return habits


    def get_habits_by_period(self, period: str):
        self.cursor.execute('SELECT * FROM habits WHERE period=?', (period,))
        rows = self.cursor.fetchall()
        habits = []
        for row in rows:
            id, name, _, created_at = row
            self.cursor.execute(
                'SELECT completed_at FROM completions WHERE habit_id=?',
                (id,)
            )
            completed_at_rows = self.cursor.fetchall()
            
class HabitTracker:
    def __init__(self):
        self.habits = []
        self.db = HabitDB()
        self.db.establish_a_connection()

    def create_habit(self, name: str, period: str):
        self.db.create_habit(name, period)

    def delete_habit(self, name: str):
        self.db.delete_habit(name)

    def get_habits(self):
        return self.db.get_habits()

    def get_habits_by_period(self, period: str):
        return self.db.get_habits_by_period(period)

    def get_longest_streak(self):
        longest_streak = 0
        longest_streak_habit = None
        for habit in self.habits:
            streak = habit.get_streak()
            if streak > longest_streak:
                longest_streak = streak
                longest_streak_habit = habit
        return longest_streak_habit

    def get_longest_streak_by_habit(self, name: str):
        for habit in self.habits:
            if habit.name == name:
                return habit.get_streak()
        return 0

    def mark_complete(self, name: str):
        self.db.mark_complete(name)

    def mark_incomplete(self, name: str):
        self.db.mark_incomplete(name)

    def get_longest_streak_by_habit(self, name: str):
        for habit in self.habits:
            if habit.name == name:
                return habit.get_streak()
        return 0

    def mark_complete(self, name: str):
        self.db.mark_complete(name)

    def mark_incomplete(self, name: str):
        self.db.mark_incomplete(name)

My new habit.py file:

from datetime import datetime



class Habit:
    def __init__(self, name: str, period: str):
        self.name = name
        self.period = period
        self.created_at = datetime.now()
        self.completed_at = []
    
    def mark_complete(self):
        self.completed_at.append(datetime.now())
        
    def mark_incomplete(self):
        self.completed_at.pop()
    
    def get_streak(self):
        if not self.completed_at:
            return 0
        current_streak = 1
        for i in range(1, len(self.completed_at)):
            if self.completed_at[i] - self.completed_at[i-1] == self.period:
                current_streak += 1
            else:
                break
        return current_streak

class HabitTracker:
    def __init__(self):
        self.habits = []
    
    def create_habit(self, name: str, period: str):
        new_habit = Habit(name, period)
        self.habits.append(new_habit)
    
    def delete_habit(self, name: str):
        for i, habit in enumerate(self.habits):
            if habit.name == name:
                del self.habits[i]
                break
    
    def get_habits(self):
        return self.habits
    
    def get_habits_by_period(self, period: str):
        return [habit for habit in self.habits if habit.period == period]
    
    def get_longest_streak(self):
        longest_streak = 0
        longest_streak_habit = None
        for habit in self.habits:
            streak = habit.get_streak()
            if streak > longest_streak:
                longest_streak = streak
                longest_streak_habit = habit
        return longest_streak_habit
    
    def get_longest_streak_by_habit(self, name: str):
        for habit in self.habits:
            if habit.name == name:
                return habit.get_streak()
        return 0
    
    def mark_complete(self, name: str):
        for habit in self.habits:
            if habit.name == name:
                habit.mark_complete()
                break
            
    def mark_incomplete(self, name: str):
        for i, completed_at in enumerate(self.completed_at):
            if self.name == name:
                del self.completed_at[i] 
            
    def relationships(self):
        for habit in self.habits:
            print(habit.name, habit.get_streak()) 
            
        
            

Perhaps this go round fixes the database issues