add export to json
This commit is contained in:
Thomas Peetz
+98
-20
@@ -1,3 +1,7 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import mariadb
|
||||
from sqlalchemy import create_engine, select, text, MetaData, join
|
||||
from sqlalchemy.orm import DeclarativeBase, relationship, sessionmaker
|
||||
@@ -30,35 +34,67 @@ class KontorDB:
|
||||
__session__ = sessionmaker(bind=engine)
|
||||
self.session = __session__()
|
||||
|
||||
def get_table_id(self, table_name):
|
||||
result = self.session.execute(select(MetaDataTable.id).where(MetaDataTable.table_name == table_name)).scalar()
|
||||
return result
|
||||
|
||||
def get_table_names(self) -> list:
|
||||
tables = self.session.query(MetaDataTable).all()
|
||||
result = [table.table_name for table in tables]
|
||||
return result
|
||||
|
||||
def get_column_meta_data(self, table_id: str, table_name: str) -> dict:
|
||||
def get_column_meta_data(self, table_name: str, view_only=True) -> dict:
|
||||
meta_data = {}
|
||||
order = 0
|
||||
for (_, column) in self.session.query(MetaDataTable, MetaDataColumn).filter(MetaDataTable.id == MetaDataColumn.table_id).filter(MetaDataTable.table_name == table_name).filter(MetaDataColumn.is_shown == 1).all():
|
||||
meta_data[order] = {'column': column.column_name, 'label': column.column_label, 'order': column.column_order, 'ref_column': column.ref_column}
|
||||
order += 1
|
||||
if view_only:
|
||||
for (_, column) in (self.session.query(MetaDataTable, MetaDataColumn).
|
||||
filter(MetaDataTable.id == MetaDataColumn.table_id).
|
||||
filter(MetaDataTable.table_name == table_name).
|
||||
filter(MetaDataColumn.is_shown == 1).all()):
|
||||
meta_data[order] = {'column': column.column_name, 'label': column.column_label,
|
||||
'order': column.column_order, 'ref_column': column.ref_column}
|
||||
order += 1
|
||||
else:
|
||||
for (_, column) in (self.session.query(MetaDataTable, MetaDataColumn).
|
||||
filter(MetaDataTable.id == MetaDataColumn.table_id).
|
||||
filter(MetaDataTable.table_name == table_name).all()):
|
||||
meta_data[order] = {
|
||||
'column': column.column_name,
|
||||
'order': column.column_order,
|
||||
'ref_column': column.ref_column
|
||||
}
|
||||
order += 1
|
||||
return meta_data
|
||||
|
||||
def get_filters(self, table_id):
|
||||
cursor = self.db_conn.cursor()
|
||||
filters = {}
|
||||
cursor.execute(
|
||||
"SELECT column_name, filter_label from meta_data_column WHERE table_id=? AND show_filter is true",
|
||||
(table_id,))
|
||||
rows = cursor.fetchall()
|
||||
for row in rows:
|
||||
filters[row[0]] = {'label': row[1], 'widget': None}
|
||||
cursor.close()
|
||||
# print(f"retrieved {len(rows)} filters: {filters}")
|
||||
return filters
|
||||
def get_filters(self, table_name):
|
||||
_filter_map = {}
|
||||
for (_, column) in (self.session.query(MetaDataTable, MetaDataColumn).
|
||||
filter(MetaDataTable.id == MetaDataColumn.table_id).
|
||||
filter(MetaDataTable.table_name == table_name).
|
||||
filter(MetaDataColumn.show_filter == 1).all()):
|
||||
_filter_map[column.column_name] = {'label': column.filter_label, 'widget': None}
|
||||
print(f"retrieved {len(_filter_map)} filters: {_filter_map}")
|
||||
return _filter_map
|
||||
|
||||
def data(self, table, columns: dict, filters) -> list:
|
||||
data = []
|
||||
entries = []
|
||||
if len(filters) == 0:
|
||||
entries = self.session.query(table).all()
|
||||
else:
|
||||
entries = self.session.query(table).filter_by(**filters)
|
||||
for entry in entries:
|
||||
row = []
|
||||
for order in columns.keys():
|
||||
column_name = columns[order]['column']
|
||||
if str(column_name).endswith("_id"):
|
||||
ref_table = column_name[:-3]
|
||||
# print(f"{ref_table=}")
|
||||
ref = getattr(entry, ref_table)
|
||||
value = getattr(ref, "name")
|
||||
# print(f"{value=}")
|
||||
row.append(value)
|
||||
else:
|
||||
row.append(getattr(entry, column_name))
|
||||
# print(repr(row))
|
||||
data.append(row)
|
||||
return data
|
||||
|
||||
def get_data(self, table_name: str, columns: dict, where_clause: str) -> list:
|
||||
data = []
|
||||
@@ -103,3 +139,45 @@ class KontorDB:
|
||||
statement = f"SELECT {columns} FROM {table} {where_clause}"
|
||||
print(f"{statement=}")
|
||||
return statement
|
||||
|
||||
def export_db(self, export_type: str, export_file_name: str, export_table_list: list):
|
||||
print(f"export DB to {export_file_name} as {export_type}")
|
||||
db = {}
|
||||
for table in export_table_list:
|
||||
columns = self.get_column_meta_data(table, view_only=False)
|
||||
model = Base.model_lookup_by_table_name(table)
|
||||
rows = self.session.query(model).all()
|
||||
entries = []
|
||||
print(f"found {len(rows)} entries")
|
||||
print(f"found {len(columns)} columns")
|
||||
for row in rows:
|
||||
print(row)
|
||||
entry = {}
|
||||
for order in columns:
|
||||
print(columns[order])
|
||||
column_name = columns[order]['column']
|
||||
print(f"get value {column_name} from {row} of table {table}")
|
||||
try:
|
||||
value = getattr(row, column_name)
|
||||
if isinstance(value, datetime):
|
||||
entry[column_name] = str(value)
|
||||
else:
|
||||
entry[column_name] = value
|
||||
except AttributeError as error:
|
||||
print("could not get value")
|
||||
entries.append(entry)
|
||||
db[table] = entries
|
||||
export_file = Path(export_file_name)
|
||||
match export_type:
|
||||
case "JSON":
|
||||
json_dump = json.dumps(db, indent=4)
|
||||
with open(export_file_name, "w") as dump_file:
|
||||
dump_file.write(json_dump)
|
||||
case "YAML":
|
||||
export_file = Path(export_file_name)
|
||||
case "SQLite":
|
||||
export_file = Path(export_file_name)
|
||||
case _:
|
||||
print("unknown export type")
|
||||
if export_file.exists():
|
||||
print(f"{export_file} exists")
|
||||
|
||||
Reference in New Issue
Block a user