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)
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:
return self("SELECT content FROM files WHERE filename=?", (filename,)).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) -> int:
return self("SELECT COUNT(*) FROM files").iloc[0, 0]
files = Files()
for filename in files:
print(files[filename])
files['hello'] = 'world'
print(files['hello'])
print(len(files))
print('hello' in files)
for filename in files:
print(filename)
del files['hello']
for filename in files:
print(filename)