add cli app and fix relationship typos

This commit is contained in:
Thomas Peetz
2025-01-13 22:54:25 +01:00
committed by Thomas Peetz
parent 89b7b87b8c
commit 54bc17ee7d
54 changed files with 1138 additions and 5 deletions
+212
View File
@@ -0,0 +1,212 @@
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
from .base import Base
from .comic import Comic, Artist, Publisher, ComicWork, WorkType, StoryArc, Volume, Issue, TradePaperback
from .tysc import Sport, Team, Card, CardSet, Vendor, Rooster, Player, FieldPosition
from .media import MediaFile
from .metadata import MetaDataTable, MetaDataColumn
class KontorDB:
def __init__(self, db_config):
self.db_conn = mariadb.connect(
host=db_config['mariadb']['host'],
port=db_config['mariadb']['port'],
user=db_config['mariadb']['user'],
password=db_config['mariadb']['password'],
database=db_config['mariadb']['database']
)
connect_string = ('mariadb+mariadbconnector://{}:{}@{}:{}/{}'
.format(
db_config['mariadb']['user'],
db_config['mariadb']['password'],
db_config['mariadb']['host'],
db_config['mariadb']['port'],
db_config['mariadb']['database']))
# engine = create_engine(connect_string, echo=True)
engine = create_engine(connect_string)
Base.metadata.create_all(bind=engine)
__session__ = sessionmaker(bind=engine)
self.session = __session__()
self.registry = {}
self.init_registry()
def init_registry(self):
self.registry['card'] = Card
self.registry['card_set'] = CardSet
self.registry['sport'] = Sport
self.registry['team'] = Team
self.registry['field_position'] = FieldPosition
self.registry['rooster'] = Rooster
self.registry['player'] = Player
self.registry['vendor'] = Vendor
self.registry['artist'] = Artist
self.registry['publisher'] = Publisher
self.registry['comic'] = Comic
self.registry['issue'] = Issue
self.registry['story_arc'] = StoryArc
self.registry['trade_paperback'] = TradePaperback
self.registry['volume'] = Volume
self.registry['comic_work'] = ComicWork
self.registry['worktype'] = WorkType
self.registry['media_file'] = MediaFile
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_name: str, view_only=True) -> dict:
meta_data = {}
order = 0
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_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 = []
cursor = self.db_conn.cursor()
cursor.execute(self.get_statement(table_name, columns, where_clause))
rows = cursor.fetchall()
print(len(rows))
for row in rows:
# print(f"KontorDB.get_data: {row}")
data.append(list(row))
cursor.close()
# print(f"KontorDB.getData: return {len(data)}")
if table_name == 'comic' and len(where_clause) == 0:
data.clear()
comics = self.session.query(Comic).all()
for item in comics:
# print(item)
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(item, ref_table)
value = getattr(ref, "name")
# print(f"{value=}")
row.append(value)
else:
row.append(getattr(item, column_name))
# print(repr(row))
data.append(row)
return data
def get_statement(self, table: str, header: dict, where_clause):
columns = ""
for index, column in header.items():
if index > 0:
columns += ", "
columns += column['column']
if len(columns) == 0:
columns = "*"
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)
if table in self.registry:
model = self.registry[table]
else:
print(f"table {table} is not registered")
continue
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")
+18
View File
@@ -0,0 +1,18 @@
from sqlalchemy.orm import DeclarativeBase, relationship, sessionmaker, declarative_base
class Base(DeclarativeBase):
pass
# class BaseModel:
#
# @classmethod
# def model_lookup_by_table_name(cls, table_name):
# registry_instance = getattr(cls, "registry")
# for mapper_ in registry_instance.mappers:
# model = mapper_.class_
# model_class_name = model.__tablename__
# if model_class_name == table_name:
# return model
# Base = declarative_base(cls=BaseModel)
+130
View File
@@ -0,0 +1,130 @@
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String
from sqlalchemy.dialects.mysql import BIT
from sqlalchemy.orm import relationship
from .base import Base
class Publisher(Base):
__tablename__ = "publisher"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(length=255), unique=True)
comics = relationship("Comic")
def __repr__(self):
return f'Publisher({self.id} {self.name})'
def __str__(self):
return self.__repr__()
class Comic(Base):
__tablename__ = 'comic'
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
title = Column(String(length=255), unique=True)
publisher_id = Column(String, ForeignKey('publisher.id'), nullable=False)
publisher = relationship("Publisher", back_populates="comics")
current_order = Column(BIT(1))
completed = Column(BIT(1))
issues = relationship("Issue")
story_arcs = relationship("StoryArc")
trade_paperbacks = relationship("TradePaperback")
volumes = relationship("Volume")
comic_works = relationship("ComicWork")
def __repr__(self):
return f'Comic({self.id} {self.version} {self.title} {self.publisher.name})'
def __str__(self):
return f'{self.title}({self.id})'
class Volume(Base):
__tablename__ = "volume"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(length=255), nullable=False)
comic_id = Column(String, ForeignKey("comic.id"), nullable=False)
comic = relationship("Comic", back_populates="volumes")
issues = relationship("Issue")
class TradePaperback(Base):
__tablename__ = "trade_paperback"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(length=255), nullable=False)
issue_start = Column(Integer)
issue_end = Column(Integer)
comic_id = Column(String, ForeignKey("comic.id"), nullable=False)
comic = relationship("Comic", back_populates="trade_paperbacks")
class StoryArc(Base):
__tablename__ = "story_arc"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(length=255), nullable=False)
comic_id = Column(String, ForeignKey("comic.id"), nullable=False)
comic = relationship("Comic", back_populates="story_arcs")
class Issue(Base):
__tablename__ = "issue"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
issue_number = Column(String(255))
in_stock = Column(BIT(1))
is_read = Column(BIT(1))
comic_id = Column(String, ForeignKey("comic.id"), nullable=False)
comic = relationship("Comic", back_populates="issues")
volume_id = Column(String, ForeignKey("volume.id"), nullable=True)
volume = relationship("Volume", back_populates="issues")
class Artist(Base):
__tablename__ = "artist"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(length=255), nullable=False)
comic_works = relationship("ComicWork")
class WorkType(Base):
__tablename__ = "worktype"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(length=255), nullable=False, unique=True)
comic_works = relationship("ComicWork")
class ComicWork(Base):
__tablename__ = "comic_work"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
comic_id = Column(String, ForeignKey("comic.id"), nullable=False)
comic = relationship("Comic", back_populates="comic_works")
artist_id = Column(String, ForeignKey("artist.id"), nullable=False)
artist = relationship("Artist", back_populates="comic_works")
work_type_id = Column(String, ForeignKey("worktype.id"), nullable=False)
work_type = relationship("WorkType", back_populates="comic_works")
+25
View File
@@ -0,0 +1,25 @@
from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.dialects.mysql import BIT
from database.base import Base
class MediaFile(Base):
__tablename__ = 'media_file'
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
cloud_link = Column(String(255))
file_name = Column(String(255))
path = Column(String(255))
review = Column(BIT(1))
title = Column(String(255))
url = Column(String(255))
should_download = Column(BIT(1))
def __repr__(self):
return f'MediaFile({self.id} {self.title} {self.title})'
def __str__(self):
return f'{self.title}({self.id})'
+49
View File
@@ -0,0 +1,49 @@
from sqlalchemy import Column, String, ForeignKey, DateTime, Integer, Boolean
from sqlalchemy.dialects.mysql import BIT
from sqlalchemy.orm import relationship
from database import Base
class MetaDataTable(Base):
__tablename__ = 'meta_data_table'
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
table_name = Column(String(255), unique=True)
table_columns = relationship("MetaDataColumn")
def __repr__(self):
return f'MetaDataTable({self.id} {self.table_name})'
def __str__(self):
return f'{self.table_name}({self.id})'
class MetaDataColumn(Base):
__tablename__ = 'meta_data_column'
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
column_modifier = Column(String(255), nullable=True)
column_name = Column(String(255))
column_order = Column(Integer)
column_sync_name = Column(String(255))
column_type = Column(String(255))
table_id = Column(String, ForeignKey('meta_data_table.id'))
table = relationship("MetaDataTable", back_populates="table_columns")
column_label = Column(String(255))
filter_label = Column(String(255))
is_shown = Column(BIT(1))
show_filter = Column(BIT(1))
ref_column = Column(String, nullable=True)
def __repr__(self):
if self.column_name is None:
return f'MetaDataColumn({self.id} {self.table.table_name}.__)'
else:
return f'MetaDataColumn({self.id} {self.table.table_name}.{self.column_name})'
def __str__(self):
return f'{self.column_name}({self.id})'
+131
View File
@@ -0,0 +1,131 @@
from sqlalchemy import Column, DateTime, Integer, String, ForeignKey, UniqueConstraint
from sqlalchemy.dialects.mysql import BIT
from sqlalchemy.orm import relationship
from .base import Base
class Sport(Base):
__tablename__ = "sport"
__table_args__ = (
UniqueConstraint("name"),
)
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(255), nullable=False, index=True, unique=True)
teams = relationship("Team")
positions = relationship("FieldPosition")
class Team(Base):
__tablename__ = "team"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(255), nullable=False, index=True, unique=True)
short_name = Column(String(255), nullable=False, )
sport_id = Column(String, ForeignKey("sport.id"), nullable=False)
sport = relationship("Sport", back_populates="teams")
roosters = relationship("Rooster")
class FieldPosition(Base):
__tablename__ = "field_position"
__table_args__ = (
UniqueConstraint("name", "sport_id"),
UniqueConstraint("short_name", "sport_id"),
)
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(255), nullable=False, index=True)
short_name = Column(String(255), nullable=False)
sport_id = Column(String, ForeignKey("sport.id"), nullable=False, index=True)
sport = relationship("Sport", back_populates="positions")
roosters = relationship("Rooster")
class Player(Base):
__tablename__ = "player"
__table_args__ = (
UniqueConstraint("first_name", "last_name"),
)
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
first_name = Column(String(255), nullable=False, index=True)
last_name = Column(String(255), nullable=False, index=True)
roosters = relationship("Rooster")
def get_full_name(self) -> str:
return f"{self.last_name}, {self.first_name}"
class Rooster(Base):
__tablename__ = "rooster"
__table_args__ = (
UniqueConstraint("year", "team_id", "player_id", "position_id"),
)
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
year = Column(Integer)
team_id = Column(String, ForeignKey("team.id"), nullable=False, index=True)
team = relationship("Team", back_populates="roosters")
player_id = Column(String, ForeignKey("player.id"), nullable=False, index=True)
player = relationship("Player", back_populates="roosters")
position_id = Column(String, ForeignKey("field_position.id"), nullable=False, index=True)
position = relationship("FieldPosition", back_populates="roosters")
cards = relationship("Card")
class Vendor(Base):
__tablename__ = "vendor"
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(255), nullable=False, unique=True, index=True)
card_sets = relationship("CardSet")
cards = relationship("Card")
class CardSet(Base):
__tablename__ = "card_set"
__table_args__ = (
UniqueConstraint("name", "vendor_id"),
)
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
name = Column(String(255), index=True)
parallel_set = Column(BIT(1))
insert_set = Column(BIT(1))
vendor_id = Column(String, ForeignKey("vendor.id"), nullable=False, index=True)
vendor = relationship("Vendor", back_populates="card_sets")
cards = relationship("Card")
class Card(Base):
__tablename__ = "card"
__table_args__ = (
UniqueConstraint("card_number", "year", "vendor_id", "card_set_id"),
)
id = Column(String, primary_key=True)
created_date = Column(DateTime)
last_modified_date = Column(DateTime)
version = Column(Integer)
card_number = Column(Integer, index=True)
year = Column(Integer, index=True)
card_set_id = Column(String, ForeignKey("card_set.id"), nullable=False)
card_set = relationship("CardSet", back_populates="cards")
rooster_id = Column(String, ForeignKey("rooster.id"), nullable=False)
rooster = relationship("Rooster", back_populates="cards")
vendor_id = Column(String, ForeignKey("vendor.id"), nullable=False)
vendor = relationship("Vendor", back_populates="cards")