Code to scan filesystem, store every path into sqlite and update only when something has changed

Question

I have made a program that does what the title explains. It also hashes the file to check for changes. I will be doing this for a project to keep track of files that has been changed when doing different things in that certain filetree. I need this to be done as an sqlite/python app to make information easily available regarding the files of interest. This will later be used in an TKinter GUI app and the Parent column contains the rowid for its parent folder to make it work with TKinter. I will make it able to use the Tags column for searching within the app as i progress.

I have been programming for a few years, but not much, I feel a lack in confidence and therefore I am throwing my code out here to get some review. I need to get better at OOP, efficiency and my logical sense needs an update from some bright minds. Please see the full code below, it is two files: filesystemScanner.py:

import os
import hashlib

from pathlib import Path
from mydatabasemanagerfile import MyDatabaseManager

path = "C:\\Users\Skaft\\Documents\\GitHub"
isFolder = True
fullList = []
currhash = ""
db_name = "parenttest.db"
table_name = "Windows10"

my_current_db = MyDatabaseManager(db_name, table_name)


#####################################################################
# copied from https://stackoverflow.com/questions/24937495/
# how-can-i-calculate-a-hash-for-a-filesystem-directory-using-python
#####################################################################
def md5_update_from_dir(directory, my_hash):
    assert Path(directory).is_dir()
    for hash_path in sorted(Path(directory).iterdir(), key=lambda p: str(p).lower()):
        my_hash.update(hash_path.name.encode())
        if hash_path.is_file():
            with open(hash_path, "rb") as f:
                for chunk in iter(lambda: f.read(4096), b""):
                    my_hash.update(chunk)
        elif hash_path.is_dir():
            my_hash = md5_update_from_dir(hash_path, my_hash)
    return my_hash


def md5_dir(directory):
    return md5_update_from_dir(directory, hashlib.md5()).hexdigest()


####################################################
# End of copy
####################################################


def file_hasher(hashing_file):
    try:
        if not isFolder:
            with open(hashing_file, "rb") as current:
                readfile = current.read()
                current_hash = hashlib.md5(readfile).hexdigest()
                return current_hash
    except PermissionError:
        current_hash = "Could not hash due to permission"
        return current_hash


for (dirpath, dirnames, filenames) in os.walk(path):
    isFolder = True
    full_dirpath = dirpath + "\\"
    fullList.append([full_dirpath, md5_dir(full_dirpath), isFolder])
    for file in filenames:
        file_path = dirpath + "\\" + file
        isFolder = False
        fullList.append([file_path, file_hasher(file_path), isFolder])

my_current_db.initialize_db()
for file_info in fullList:
    my_current_db.my_updater(file_info[0], file_info[1], file_info[2])

my_current_db.inserting_parentID()
my_current_db.inserting_suffix()
my_current_db.close_db()

And mydatabasemanagerfile.py

import sqlite3
import datetime
import pathlib


class MyDatabaseManager:
    def __init__(self, db_name, table_name):
        self.db_name = db_name
        self.table_name = table_name
        self.conn = sqlite3.connect(self.db_name)
        self.cur = self.conn.cursor()

    def initialize_db(self):
        """Function to open database, create if not exists"""
        print("Opened database successfully")

        # drop_if_exist = "DROP TABLE IF EXISTS {}".format(self.table_name)
        # self.cur.execute(drop_if_exist)

        create_table = '''CREATE TABLE IF NOT EXISTS {}
                ("Path"             TEXT NOT NULL,
                "Information"       TEXT,
                "Tags"              TEXT,
                "Parent"            INTEGER,
                "isFolder"          INTEGER,
                "fileExt"           TEXT,
                "MD5"               TEXT,
                "Updated"           TEXT,
                "See also"          TEXT
                )'''.format(self.table_name)
        self.cur.execute(create_table)
        print("Table created successfully")

    def my_updater(self, path, md5, isFolder):
        query_path = '''SELECT rowid, Path FROM {} WHERE Path = "{}"'''.format(self.table_name, path)

        if self.conn.execute(query_path).fetchone() is None:  # Test if returns None.
            result_path = "Please add me"  # Placeholder to specify that it needs to be inserted, it does not exist
            result_hash = "Please add me"
        else:

            result_path = self.conn.execute(query_path).fetchone()[1]
            rowid = self.conn.execute(query_path).fetchone()[0]
            query_hash = '''SELECT MD5 FROM {} WHERE rowid = "{}"'''.format(self.table_name, rowid)

            result_hash = self.conn.execute(query_hash).fetchone()[0]
        current_time = datetime.datetime.now()

        if path == result_path and md5 != result_hash:
            query = '''UPDATE {} SET MD5=?, Updated=? WHERE rowid = "{}"'''.format(self.table_name, rowid)
            self.cur.execute(query, (md5, current_time))
        elif path != result_path:
            query = '''INSERT INTO {} (Path, MD5, isFolder, Updated) VALUES (?, ?, ?, ?)'''.format(self.table_name)
            self.cur.execute(query, (path, md5, isFolder, current_time))
        else:
            #print("Path: ", path, "\n", "result_path: ", result_path, "\n", "md5      : ", md5, "\n", "result_hash: ",
            #      result_hash, "\n")
            pass
        self.conn.commit()

    def inserting_parentID(self):
        query = '''SELECT rowid, Path FROM {}'''.format(self.table_name)
        all_paths = self.cur.execute(query).fetchall()

        for rowid, path in all_paths:
            parent_path_split = path.split("\\")
            if rowid == 1:
                continue
            if path[-1] == "\\":
                parent_path_joined = "\\".join(parent_path_split[:-2]) + "\\"
                parent_path_joinedads = "\\".join(parent_path_split[:-2]) + "\\"
            else:
                parent_path_joined = "\\".join(parent_path_split[:-1]) + "\\"

            parent_query = '''SELECT rowid FROM {} WHERE Path = "{}"'''.format(self.table_name, parent_path_joined)
            parent_paths = self.cur.execute(parent_query).fetchone()

            update_parentID = '''UPDATE {} SET Parent=? WHERE rowid = "{}"'''.format(self.table_name, rowid)
            self.cur.execute(update_parentID, (parent_paths))
            self.conn.commit()

    def inserting_suffix(self):
        query = '''SELECT rowid, Path FROM {}'''.format(self.table_name)
        all_paths = self.cur.execute(query).fetchall()

        for rowid, path in all_paths:
            ext = pathlib.Path(path).suffix
            if ext == "":
                continue
            update_ext_query = '''UPDATE {} SET fileExt=? WHERE rowid = "{}"'''.format(self.table_name, rowid)
            self.cur.execute(update_ext_query, (ext,))
            self.conn.commit()



    def close_db(self):
        self.conn.close()

Any feedback is appreciated to get a more professional and standardized result. I find this a good way to learn, to view and discuss improvement on my own code.

Answer 1

Avoid hard-coding file paths like this:

path = "C:\\Users\Skaft\\Documents\\GitHub"

That path should be parametric.

Variable names like isFolder should be is_folder (snake_case) via PEP8.

Run-time assertions of the kind seen here:

assert Path(directory).is_dir()

can be trivially disabled in Python and are not appropriate for production code. Raise an exception instead.

You say you have files of multiple GB. You've done the right thing here:

            for chunk in iter(lambda: f.read(4096), b""):
                my_hash.update(chunk)

and reimplemented it, incorrectly, here:

            current_hash = hashlib.md5(readfile).hexdigest()

Consider factoring out a single file-hashing method that uses the chunked approach to go easy on your RAM.

Do not use string interpolation to add query parameters:

    query_path = '''SELECT rowid, Path FROM {} WHERE Path = "{}"'''.format(self.table_name, path)

That's the classic vector for injection attacks. Instead use ? parameter substitution; read the documentation.

Don't use in-band error signalling, like here:

except PermissionError:
    current_hash = "Could not hash due to permission"
    return current_hash

If both a hash and your error message are a string, it's problematic for a caller to tell them apart. Just let the permission error exception fall through and catch it on the calling side.