52 lines
1.0 KiB
Python
52 lines
1.0 KiB
Python
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import zipfile
|
|
from json import loads as json_loads
|
|
|
|
|
|
class Loader:
|
|
def load(self, file: str):
|
|
raise NotImplementedError()
|
|
|
|
def get_entry(self) -> object:
|
|
raise NotImplementedError()
|
|
|
|
|
|
class JSONLoader(Loader):
|
|
data = None
|
|
|
|
def load(self, file: str):
|
|
self.data = json.load(open(file))
|
|
|
|
def get_entry(self) -> dict:
|
|
for entry in self.data:
|
|
yield entry
|
|
|
|
|
|
class SQLiteLoader(Loader):
|
|
conn = None
|
|
|
|
def load(self, file: str):
|
|
self.conn = sqlite3.connect(file)
|
|
|
|
def get_entry(self) -> dict:
|
|
cursor = self.conn.cursor()
|
|
cursor.execute("SELECT * FROM log_entry")
|
|
for seq, timestamp, json in cursor.fetchall():
|
|
yield json_loads(json)
|
|
|
|
|
|
class ZipSQLiteLoader(SQLiteLoader):
|
|
def load(self, file: str):
|
|
with zipfile.ZipFile(file, "r") as zipped_log, tempfile.TemporaryDirectory() as tmp:
|
|
zipped_log.extract("instance_log.sqlite", path=tmp)
|
|
super(ZipSQLiteLoader, self).load("{dir}/instance_log.sqlite".format(dir=tmp))
|
|
|
|
|
|
LOADERS = {
|
|
"json": JSONLoader,
|
|
"sqlite": SQLiteLoader,
|
|
"zip": ZipSQLiteLoader
|
|
}
|