roberteam/lewangoalski
6
2
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: orm support for db calls

Browse Source 
sample call from debugger_halt():
`getDb().simple_select_all("mecze", id_meczu=1)`
This commit is contained in:
sherl
2025-05-30 01:45:35 +02:00
parent 72141768d4
commit 56f90efe40
1 changed files with 77 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
FlaskWebProject/FlaskWebProject/lewy_db.py
View File

@@ -1,5 +1,6 @@
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
@@ -12,9 +13,14 @@ 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.db = self.initDB(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
@@ -48,6 +54,9 @@ class baza():
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):
@@ -59,6 +68,9 @@ class baza():
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)
@@ -75,6 +87,9 @@ class baza():
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)
@@ -93,6 +108,9 @@ class baza():
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)
@@ -101,6 +119,9 @@ class baza():
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)
@@ -117,25 +138,64 @@ class baza():
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 simple_select_all(self, type, **kwargs):
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:
self.session.rollback()
self.session.close()
self.refresh_session()
return return_val
return wrapper
@exit_gracefully
def simple_select_all(self, entity_type, **kwargs):
"""
Użycie:
simple_select_all(ldb.sportowcy, zewnetrzne_id_zawodnika="").fetchall()
simple_select_all(ldb.sportowcy, zewnetrzne_id_zawodnika="MVC8zHZD")
simple_select_all("sportowcy", id_zawodnika=1)
https://stackoverflow.com/a/75316945
Did they make it harder to query dynamically on purpose? ~Frank 19.11.2023
"""
with self.db.engine.begin() as conn:
return conn.execute(
select(type).
filter_by(**kwargs)
).scalars()
if not isinstance(entity_type, str):
entity_type = entity_type.__name__
results = (
self.session.
query(self.entities[entity_type]).
filter_by(**kwargs).
all()
)
return results
@exit_gracefully
def simple_insert_one(self, type, **kwargs):
"""
Użycie:
@@ -151,6 +211,7 @@ class baza():
return 0
return 1
@exit_gracefully
def simple_insert_many(self, objs_list):
"""
Użycie:
@@ -165,6 +226,7 @@ class baza():
return 0
return 1
@exit_gracefully
def sample_data_init(self, override_safety_check=False):
"""
Użycie:
@@ -176,8 +238,13 @@ class baza():
sample_data_init(override_safety_check=True)
Metoda głównie używana do testów.
"""
is_table_empty = self.simple_select_all(sportowcy, zewnetrzne_id_zawodnika="MVC8zHZD").one() is None
if not is_table_empty and not override_safety_check:
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: