391 lines
17 KiB
Python
391 lines
17 KiB
Python
from datetime import datetime
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from functools import wraps
|
|
from sqlalchemy import ForeignKey, select, insert, update
|
|
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase, Session, relationship
|
|
from typing import List
|
|
import time
|
|
import toml
|
|
import traceback
|
|
|
|
global db
|
|
|
|
class c:
|
|
HEADER = '\033[95m'
|
|
OKBLUE = '\033[94m'
|
|
OKCYAN = '\033[96m'
|
|
OKGREEN = '\033[92m'
|
|
WARNING = '\033[93m'
|
|
FAIL = '\033[91m'
|
|
ENDC = '\033[0m'
|
|
BOLD = '\033[1m'
|
|
UNDERLINE = '\033[4m'
|
|
ENDL = '\n'
|
|
|
|
class baza():
|
|
|
|
# global sportowcy, trofea, sportowcy_w_meczach, statystyki_sportowcow, kluby, mecze
|
|
|
|
db = None
|
|
entities = {}
|
|
session = None
|
|
app = None
|
|
|
|
def __init__(self, app, config):
|
|
self.app = app
|
|
self.db = self.initDB(self.app, config)
|
|
self.refresh_session()
|
|
|
|
def initDB(self, app, config):
|
|
global sportowcy, trofea, sportowcy_w_meczach, statystyki_sportowcow, kluby, mecze
|
|
tnp = config['general']['db_prefix'] + "_lewangoalski_"
|
|
|
|
class Base(DeclarativeBase):
|
|
pass
|
|
|
|
db = SQLAlchemy(app, model_class=Base)
|
|
|
|
class sportowcy(Base):
|
|
__tablename__ = tnp + "sportowcy"
|
|
id_zawodnika: Mapped[ int] = mapped_column(primary_key=True, autoincrement=True)
|
|
zewnetrzne_id_zawodnika: Mapped[ str] = mapped_column(unique=True)
|
|
imie: Mapped[ str] = mapped_column()
|
|
nazwisko: Mapped[ str] = mapped_column()
|
|
data_urodzenia: Mapped[ str] = mapped_column()
|
|
czy_aktywny: Mapped[ bool] = mapped_column()
|
|
klub_id: Mapped[ List[str]] = mapped_column(ForeignKey(f"{tnp}kluby.id_klubu"), nullable=True)
|
|
klub: Mapped[ List["kluby"]] = relationship(back_populates="sportowcy_w_klubie", foreign_keys=[klub_id])
|
|
narodowosc: Mapped[ str] = mapped_column()
|
|
ilosc_trofeow: Mapped[ int] = mapped_column()
|
|
ostatnie_trofeum_id: Mapped[ int] = mapped_column(ForeignKey(f"{tnp}trofea.id_trofeum"), nullable=True)
|
|
ostatnie_trofeum: Mapped[ "trofea"] = relationship(back_populates="zawodnik", foreign_keys=[ostatnie_trofeum_id])
|
|
pierwszy_mecz_id: Mapped[ int] = mapped_column(ForeignKey(f"{tnp}mecze.id_meczu"), nullable=True)
|
|
pierwszy_mecz: Mapped[ "mecze"] = relationship()
|
|
wycena: Mapped[ int] = mapped_column()
|
|
ostatni_gol_dla_id: Mapped[ str] = mapped_column(ForeignKey(f"{tnp}kluby.id_klubu"), nullable=True)
|
|
ostatni_gol_dla: Mapped[ "kluby"] = relationship(back_populates="sportowcy_ostatni_gol", foreign_keys=[ostatni_gol_dla_id])
|
|
statystyki_id: Mapped[ List[int]] = mapped_column(ForeignKey(f"{tnp}statystyki_sportowcow.id_statystyki"), nullable=True)
|
|
statystyki: Mapped[List["statystyki_sportowcow"]] = relationship(back_populates="sportowiec")
|
|
trofea: Mapped[ List["trofea"]] = relationship(back_populates="zawodnik", foreign_keys="[trofea.id_zawodnika]")
|
|
|
|
def __repr__(self):
|
|
return f"<Sportowiec #{self.id_zawodnika} ({self.imie} {self.nazwisko})>"
|
|
|
|
# Co było pierwsze, jajko czy kura? https://docs.sqlalchemy.org/en/20/orm/relationship_persistence.html#rows-that-point-to-themselves-mutually-dependent-rows
|
|
# Kiepskie rozwiązanie, ale jednak działające: pozwolić obcym kluczom na bycie "null"
|
|
class trofea(Base):
|
|
__tablename__ = tnp + "trofea"
|
|
id_trofeum: Mapped[ int] = mapped_column(primary_key=True, autoincrement=True)
|
|
id_zawodnika: Mapped[ int] = mapped_column(ForeignKey(f"{tnp}sportowcy.id_zawodnika", name="fk_zawodnik"), nullable=True)
|
|
zawodnik: Mapped[ "sportowcy"] = relationship(back_populates="trofea", foreign_keys=[id_zawodnika], post_update=True)
|
|
nazwa: Mapped[ str] = mapped_column()
|
|
sezon: Mapped[ str] = mapped_column()
|
|
rok: Mapped[ int] = mapped_column()
|
|
|
|
def __repr__(self):
|
|
return f"<Trofeum #{self.id_trofeum} ({self.nazwa})>"
|
|
|
|
class sportowcy_w_meczach(Base):
|
|
__tablename__ = tnp + "sportowcy_w_meczach"
|
|
id_rekordu: Mapped[ int] = mapped_column(primary_key=True, autoincrement=True)
|
|
id_zawodnika: Mapped[ int] = mapped_column(ForeignKey(f"{tnp}sportowcy.id_zawodnika"))
|
|
zawodnik: Mapped[ "sportowcy"] = relationship()
|
|
zewnetrzne_id_meczu: Mapped[ str] = mapped_column(ForeignKey(f"{tnp}mecze.zewnetrzne_id_meczu"))
|
|
czas_gry: Mapped[ int] = mapped_column()
|
|
goli: Mapped[ int] = mapped_column()
|
|
asyst: Mapped[ int] = mapped_column()
|
|
interwencje_bramkarza: Mapped[ int] = mapped_column()
|
|
suma_interwencji_na_bramke: Mapped[ int] = mapped_column()
|
|
zolte_kartki: Mapped[ int] = mapped_column()
|
|
czerwone_kartki: Mapped[ int] = mapped_column()
|
|
wygrana: Mapped[ int] = mapped_column()
|
|
wynik: Mapped[ float] = mapped_column()
|
|
|
|
def __repr__(self):
|
|
return f"<Sportowiec #{self.id_zawodnika} ({self.imie} {self.nazwisko})>"
|
|
|
|
class statystyki_sportowcow(Base):
|
|
__tablename__ = tnp + "statystyki_sportowcow"
|
|
id_statystyki: Mapped[ int] = mapped_column(primary_key=True, autoincrement=True)
|
|
sportowiec: Mapped[ "sportowcy"] = relationship(back_populates="statystyki")
|
|
ostatni_mecz: Mapped[ int] = mapped_column(ForeignKey(f"{tnp}mecze.id_meczu"))
|
|
ilosc_wystapien: Mapped[ int] = mapped_column()
|
|
minut_gry: Mapped[ int] = mapped_column()
|
|
gier_sum: Mapped[ int] = mapped_column()
|
|
goli_sum: Mapped[ int] = mapped_column()
|
|
asyst_sum: Mapped[ int] = mapped_column()
|
|
interwencji_sum: Mapped[ int] = mapped_column()
|
|
nieobronionych_interwencji_sum: Mapped[ int] = mapped_column()
|
|
zoltych_kartek_sum: Mapped[ int] = mapped_column()
|
|
czerwonych_kartek_sum: Mapped[ int] = mapped_column()
|
|
wygranych_sum: Mapped[ int] = mapped_column()
|
|
wynik_sum: Mapped[ int] = mapped_column()
|
|
meczow_do_wynikow_sum: Mapped[ int] = mapped_column()
|
|
|
|
def __repr__(self):
|
|
return f"<Statystyka #{self.id_statystyki} ({self.sportowiec.imie} {self.sportowiec.nazwisko})>"
|
|
|
|
class kluby(Base):
|
|
__tablename__ = tnp + "kluby"
|
|
id_klubu: Mapped[ str] = mapped_column(primary_key=True)
|
|
pelna_nazwa: Mapped[ str] = mapped_column()
|
|
skrocona_nazwa: Mapped[ str] = mapped_column()
|
|
sportowcy_w_klubie: Mapped[ List["sportowcy"]] = relationship(back_populates="klub", foreign_keys="[sportowcy.klub_id]")
|
|
sportowcy_ostatni_gol: Mapped[ "sportowcy"] = relationship(back_populates="ostatni_gol_dla", foreign_keys="[sportowcy.ostatni_gol_dla_id]")
|
|
|
|
def __repr__(self):
|
|
return f"<Klub #{self.id_klubu} ({self.skrocona_nazwa})>"
|
|
|
|
class mecze(Base):
|
|
__tablename__ = tnp + "mecze"
|
|
id_meczu: Mapped[ int] = mapped_column(primary_key=True, autoincrement=True)
|
|
zewnetrzne_id_meczu: Mapped[ str] = mapped_column(unique=True)
|
|
data: Mapped[ datetime] = mapped_column()
|
|
gospodarze_id: Mapped[ str] = mapped_column(ForeignKey(f"{tnp}kluby.id_klubu"))
|
|
gospodarze: Mapped[ "kluby"] = relationship(foreign_keys=[gospodarze_id])
|
|
goscie_id: Mapped[ str] = mapped_column(ForeignKey(f"{tnp}kluby.id_klubu"))
|
|
goscie: Mapped[ "kluby"] = relationship(foreign_keys=[goscie_id])
|
|
gosp_wynik: Mapped[ int] = mapped_column()
|
|
gosc_wynik: Mapped[ int] = mapped_column()
|
|
sezon: Mapped[ str] = mapped_column()
|
|
nazwa_turnieju: Mapped[ str] = mapped_column()
|
|
skrocona_nazwa_turnieju: Mapped[ str] = mapped_column()
|
|
flaga: Mapped[ int] = mapped_column()
|
|
|
|
def __repr__(self):
|
|
return f"<Mecz #{self.id_meczu} ({self.zewnetrzne_id_meczu}, {self.gospodarze.skrocona_nazwa} vs. {self.goscie.skrocona_nazwa})>"
|
|
|
|
self.entities = {
|
|
'sportowcy': sportowcy,
|
|
'trofea': trofea,
|
|
'sportowcy_w_meczach': sportowcy_w_meczach,
|
|
'statystyki_sportowcow': statystyki_sportowcow,
|
|
'kluby': kluby,
|
|
'mecze': mecze
|
|
}
|
|
|
|
return db
|
|
|
|
def create_all(self):
|
|
self.db.create_all()
|
|
|
|
def refresh_session(self):
|
|
with self.app.app_context():
|
|
self.session = Session(self.db.engine)
|
|
|
|
def exit_gracefully(func):
|
|
@wraps(func)
|
|
def wrapper(self, *args, **kwargs):
|
|
return_val = None
|
|
try:
|
|
return_val = func(self, *args, **kwargs)
|
|
except:
|
|
print(f"{c.FAIL}"
|
|
f"Wystąpił błąd podczas wykonywania zapytania SQL:"
|
|
f"{c.ENDC}"
|
|
"\n"
|
|
f"{traceback.format_exc()}")
|
|
self.session.rollback()
|
|
self.session.close()
|
|
self.refresh_session()
|
|
return return_val
|
|
return wrapper
|
|
|
|
def str_to_column(self, string: str):
|
|
"""
|
|
Zamienia tekstowy zapis "tabela.kolumna"
|
|
na obiekt modelu bazy (ze zmiennej self.entities).
|
|
Zwraca None jeśli takowego nie znajdzie.
|
|
Zamiennik dla niepożądanego eval().
|
|
|
|
:param string: Zapis tekstowy
|
|
:type string: str
|
|
"""
|
|
table_str = string[:string.find('.')]
|
|
column_str = string[string.find('.') + 1:]
|
|
if hasattr(self.entities[table_str], column_str):
|
|
return getattr(self.entities[table_str], column_str)
|
|
return None
|
|
|
|
@exit_gracefully
|
|
def simple_select_all(self, entity_type, **kwargs):
|
|
"""
|
|
Użycie:
|
|
simple_select_all(ldb.sportowcy, zewnetrzne_id_zawodnika="MVC8zHZD")
|
|
simple_select_all("sportowcy", id_zawodnika=1)
|
|
simple_select_all("kluby", ..., LIMIT=5, ORDER_BY="kluby.skrocona_nazwa")
|
|
|
|
https://stackoverflow.com/a/75316945
|
|
Did they make it harder to query dynamically on purpose? ~Frank 19.11.2023
|
|
"""
|
|
|
|
if not isinstance(entity_type, str):
|
|
entity_type = entity_type.__name__
|
|
|
|
# Save special arguments received with kwargs,
|
|
# that are meant for SQL operations to special_args,
|
|
# and delete from the rest, that will be passed
|
|
# directly to filter_by().
|
|
# They will not be passed as search query, but serve
|
|
# as an additional search parameter.
|
|
special_keywords = ("ORDER_BY", "ORDER_BY_DESC", "LIMIT")
|
|
special_args = {}
|
|
|
|
for arg in special_keywords:
|
|
if arg in kwargs:
|
|
special_args[arg] = kwargs[arg]
|
|
del kwargs[arg]
|
|
|
|
print(f"[{round(time.time())}] SELECT")
|
|
|
|
results = (
|
|
self.session.
|
|
query(self.entities[entity_type]).
|
|
filter_by(**kwargs)
|
|
)
|
|
|
|
if "ORDER_BY" in special_args:
|
|
column = self.str_to_column(special_args["ORDER_BY"])
|
|
if column is not None:
|
|
results = results.order_by(column)
|
|
|
|
if "ORDER_BY_DESC" in special_args:
|
|
column = self.str_to_column(special_args["ORDER_BY_DESC"])
|
|
if column is not None:
|
|
results = results.order_by(column.desc())
|
|
|
|
if "LIMIT" in special_args:
|
|
results = results.limit(special_args["LIMIT"])
|
|
|
|
results_objs = results.all()
|
|
print(f"[{round(time.time())}] SELECT RESULTS: {results_objs}")
|
|
|
|
return results_objs
|
|
|
|
@exit_gracefully
|
|
def simple_insert_one(self, entity_type, **kwargs):
|
|
"""
|
|
Użycie:
|
|
simple_insert_one(ldb.kluby, id_klubu="polska", pelna_nazwa="Reprezentacja Polski", skrocona_nazwa="PL")
|
|
|
|
https://docs.sqlalchemy.org/en/20/tutorial/data_insert.html
|
|
https://docs.sqlalchemy.org/en/20/orm/session_basics.html
|
|
"""
|
|
|
|
if not isinstance(entity_type, str):
|
|
entity_type = entity_type.__name__
|
|
|
|
if "id" in kwargs:
|
|
print(f"{c.FAIL}UWAGA!{c.ENDC}")
|
|
print(f"Próbujesz dodać obiekt do tabeli, który ma już identyfikator.\n"
|
|
f"To spowoduje problemy w przyszłości, gdy będziesz chciał dodać nowy obiekt do bazy bez ustawiania id na sztywno\n"
|
|
f"(id klucza głównego nie zostanie zaktualizowane w sekwencji, przez co baza będzie próbowała dodać obiekt z id już istniejącego rekordu!).\n"
|
|
f"Aby naprawić dodawanie z autoinkrementującym kluczem zobacz {c.WARNING}https://stackoverflow.com/a/8745101{c.ENDC}\n"
|
|
f"Zostałeś ostrzeżony!")
|
|
|
|
print(f"[{round(time.time())}] INSERT")
|
|
|
|
obj = self.entities[entity_type](**kwargs)
|
|
#with Session(self.db.engine) as session:
|
|
self.session.add(obj)
|
|
self.session.commit()
|
|
return 0
|
|
#return 1
|
|
|
|
@exit_gracefully
|
|
def simple_insert_many(self, objs_list):
|
|
"""
|
|
Użycie:
|
|
simple_insert_many([sportowiec_a, sportowiec_b])
|
|
|
|
https://docs.sqlalchemy.org/en/20/tutorial/data_insert.html
|
|
https://docs.sqlalchemy.org/en/20/orm/session_basics.html
|
|
"""
|
|
#with Session(self.db.engine) as session:
|
|
self.session.add_all(objs_list)
|
|
self.session.commit()
|
|
return 0
|
|
#return 1
|
|
|
|
@exit_gracefully
|
|
def sample_data_init(self, override_safety_check=False):
|
|
"""
|
|
Użycie:
|
|
sample_data_init()
|
|
Uwaga! Poniższe populuje pustą bazę danych.
|
|
Nie należy tego używać na nie-pustej bazie, ponieważ spowoduje
|
|
to wpisanie śmieciowych danych!
|
|
Jeżeli wiesz co robisz w takiej sytuacji, uruchom metodę z:
|
|
sample_data_init(override_safety_check=True)
|
|
Metoda głównie używana do testów.
|
|
"""
|
|
is_table_empty = False
|
|
try:
|
|
self.simple_select_all(self.sportowcy, zewnetrzne_id_zawodnika="MVC8zHZD")
|
|
except:
|
|
is_table_empty = True
|
|
|
|
if not is_table_empty and override_safety_check:
|
|
raise EnvironmentError("sample_data_init() ran on a non-empty database. Ignore with override_safety_check=True.")
|
|
|
|
with Session(self.db.engine) as session:
|
|
self.simple_insert_one(kluby,
|
|
id_klubu="undefined",
|
|
pelna_nazwa="Klub niezdefiniowany",
|
|
skrocona_nazwa="N/A")
|
|
self.simple_insert_one(mecze,
|
|
zewnetrzne_id_meczu="dummy_match",
|
|
data=datetime.strptime("1970-01-01 00:00:00", '%Y-%m-%d %H:%M:%S'),
|
|
gospodarze_id="undefined",
|
|
goscie_id="undefined",
|
|
gosp_wynik=0,
|
|
gosc_wynik=0,
|
|
sezon="1970/1970",
|
|
nazwa_turnieju="Nieznany turniej",
|
|
skrocona_nazwa_turnieju="N/A",
|
|
flaga=0)
|
|
self.simple_insert_one(statystyki_sportowcow,
|
|
ostatni_mecz=1,
|
|
ilosc_wystapien=0,
|
|
minut_gry=0,
|
|
gier_sum=0,
|
|
goli_sum=0,
|
|
asyst_sum=0,
|
|
interwencji_sum=0,
|
|
nieobronionych_interwencji_sum=0,
|
|
zoltych_kartek_sum=0,
|
|
czerwonych_kartek_sum=0,
|
|
wygranych_sum=0,
|
|
wynik_sum=0,
|
|
meczow_do_wynikow_sum=0)
|
|
|
|
sportowiec = sportowcy(
|
|
zewnetrzne_id_zawodnika="MVC8zHZD",
|
|
imie="Robert",
|
|
nazwisko="Lewandowski",
|
|
data_urodzenia="21.08.1988",
|
|
czy_aktywny=True,
|
|
klub_id="undefined",
|
|
narodowosc="PL",
|
|
ilosc_trofeow=0,
|
|
pierwszy_mecz_id=1,
|
|
ostatni_gol_dla_id="undefined",
|
|
statystyki_id=1,
|
|
wycena=64_940_000)
|
|
trofeum = trofea(
|
|
nazwa="Nieznane trofeum",
|
|
sezon="0000/0000",
|
|
rok=1970)
|
|
|
|
session.add(sportowiec)
|
|
session.flush()
|
|
|
|
session.add(trofeum)
|
|
session.flush()
|
|
|
|
trofeum.zawodnik = sportowiec
|
|
sportowiec.ostatnie_trofeum = trofeum
|
|
|
|
session.commit()
|
|
return 0
|
|
return 1 |