SQLite-backed Python dict

sqlite-dict.py

"""Companion to https://oatzy.github.io/2021/07/11/persistent-dict-with-sqlite.html."""

import sqlite3 as sql
from collections.abc import MutableMapping
from datetime import datetime


class DateTimeMap(MutableMapping):
    """Persistent map of when 'something' occured, backed by SQLite3.

    If the required table doesn't exist in the connection db
    it will be created, with name 'events'

    Makes use of the python builtin sqlite3 library.
    
    For values to be loaded as datetime instances,
    the sql connection needs to be created with
    
    >>> sql.connect(path, detect_types=sql.PARSE_DECLTYPES)
    """

    def __init__(self, connection: sql.Connection):
        self.connection = connection
        self.connection.execute("CREATE TABLE IF NOT EXISTS 'events' (id TEXT PRIMARY KEY, timestamp TIMESTAMP)")

    def __getitem__(self, item_id: str) -> datetime:
        item = self.connection.execute("SELECT path FROM 'events' WHERE id=?", (item_id,)).fetchone()
        if item is None:
            raise KeyError(item_id)
        # 'item' is a single entry tuple
        return item[0]

    def __setitem__(self, item_id: str, timestamp: datetime):
        self.connection.execute("REPLACE INTO 'events' VALUES(?,?)", (item_id, timestamp))
        self.connection.commit()

    def __delitem__(self, item_id: str):
        _ = self[item_id]  # KeyError if id doesn't exist
        self.connection.execute("DELETE FROM 'events' WHERE id=?", (item_id,))
        self.connection.commit()

    def __iter__(self):
        for item in self.connection.execute("SELECT id FROM 'events'"):
            yield item[0]

    def __len__(self):
        res = self.connection.execute("SELECT COUNT(id) FROM 'events'").fetchone()
        return res[0]

    def clear(self):
        """Bulk delete all entries (rows)."""
        # the default implementation for MutableMapping is iterative
        self.connection.execute("DELETE FROM 'events'")
        self.connection.commit()