wiki/sqlite3/Files

# files.py
import pandas as pd

class Files:

    def __init__(self, db='files.sqlite'):
        import sqlite3
        if not db.endswith('.sqlite'): db += '.sqlite'
        self.conn = sqlite3.connect(db, isolation_level=None, timeout=60)
        self.conn.row_factory = sqlite3.Row
        self("PRAGMA journal_mode=WAL")
        self("CREATE TABLE IF NOT EXISTS files (filename TEXT PRIMARY KEY, content BLOB)")

    def __call__(self, *args) -> pd.DataFrame:
        return pd.DataFrame(dict(row) for row in self.conn.execute(*args))

    def __setitem__(self, filename, content) -> None:
        assert isinstance(filename, str)
        if isinstance(content, str): content = content.encode()
        assert isinstance(content, bytes)
        self("INSERT OR REPLACE INTO files VALUES (?,?)", (filename, content))

    def __getitem__(self, filename) -> bytes:
        result = self("SELECT content FROM files WHERE filename=?", (filename,))
        if result.empty: raise KeyError(filename)
        return result.content[0]

    def __delitem__(self, filename) -> None:
        self("DELETE FROM files WHERE filename=?", (filename,))

    def __iter__(self):
        df = self("SELECT filename FROM files")
        return iter([] if df.empty else df.filename)

    def __contains__(self, filename: str) -> bool:
        return not self("SELECT 1 FROM files WHERE filename=?", (filename,)).empty

    def __len__(self): return self("SELECT COUNT(*) FROM files").iloc[0, 0]
    def __enter__(self): return self
    def __exit__(self, exc_type, exc_val, exc_tb): self.close()
    def __del__(self): self.close()
    
    def close(self):
        if self.conn:
            self('PRAGMA wal_checkpoint(FULL)')
            self.conn.close()
            self.conn = None

if __name__ == "__main__":
    fs = Files()
    for filename in fs:
        print(filename, fs[filename])
    fs['hello'] = 'world'
    print(fs['hello'])
    print(len(fs))

    print('hello' in fs)
    for filename in fs:
        print(filename)

    del fs['hello']
    for filename in fs:
        print(filename)

    with Files() as fs:
        fs['world'] = '1234'
        print(fs['world'])
        print(len(fs))