67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
from enum import Enum, auto
|
|
from logging import Logger
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from db.models import registry
|
|
|
|
|
|
class ColumnEntry(Enum):
|
|
COLUMN_NAME = 'column'
|
|
COLUMN_LABEL = 'label'
|
|
COLUMN_ORDER = 'order'
|
|
COLUMN_REF_COLUMN = 'ref_column'
|
|
COLUMN_TYPE = 'type'
|
|
COLUMN_WIDGET = 'widget'
|
|
|
|
|
|
class StatusType(Enum):
|
|
UNKNOWN = auto()
|
|
FILE_NAME = auto()
|
|
FILE_ID = auto()
|
|
DUPLICATE = auto()
|
|
CLOUD_LINK = auto()
|
|
CLOUD_LINK_ID = auto()
|
|
|
|
|
|
class ExportType(Enum):
|
|
JSON = "JSON"
|
|
YAML = "YAML"
|
|
SQLITE = "SQLite"
|
|
|
|
|
|
class KontorDB:
|
|
|
|
def __init__(self, db_engine: Any, log: Logger):
|
|
self.engine = db_engine
|
|
self.log = log
|
|
|
|
def data(self, table_name: str, columns: dict, filters: dict) -> list:
|
|
data = []
|
|
__session__ = sessionmaker(self.engine)
|
|
table = registry[table_name]
|
|
with __session__() as session:
|
|
entries = []
|
|
if len(filters) == 0:
|
|
entries = session.scalars(select(table)).all()
|
|
else:
|
|
entries = session.scalars(select(table).filter_by(**filters)).all()
|
|
for entry in entries:
|
|
# self.log.info("data: %s", entry)
|
|
row = []
|
|
for order in columns.keys():
|
|
column_name = columns[order][ColumnEntry.COLUMN_NAME]
|
|
ref_column = columns[order][ColumnEntry.COLUMN_REF_COLUMN]
|
|
if str(column_name).endswith("_id"):
|
|
ref_table = column_name[:-3]
|
|
ref = getattr(entry, ref_table)
|
|
value = getattr(ref, ref_column)
|
|
row.append(value)
|
|
else:
|
|
row.append(getattr(entry, column_name))
|
|
data.append(row)
|
|
# self.log.info("data: %s", data)
|
|
return data
|