Reformatted code to pythonic standards

This commit is contained in:
chris committed 2026-05-27 08:16:35 -04:00
1 parent 31b335b7e8
commit e2b8f5497c
1 file changed
+58 -21
+58 -21
View File
@@ -6,13 +6,19 @@
from datetime import date, datetime from datetime import date, datetime
from typing import Any, Sequence, TypeVar from typing import Any, Sequence, TypeVar
from sqlalchemy import create_engine, Date, engine, func, Integer, MetaData, Row, text from sqlalchemy import (
create_engine, Date, func, Integer, MetaData,
Row, text,
)
from sqlalchemy.exc import IntegrityError, PendingRollbackError from sqlalchemy.exc import IntegrityError, PendingRollbackError
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, Session from sqlalchemy.orm import (
DeclarativeBase, Mapped, mapped_column, Session, sessionmaker,
)
from sqlalchemy_utils import create_database, database_exists from sqlalchemy_utils import create_database, database_exists
from util.environment import load_environment_variables from util.environment import load_environment_variables
project_variables = load_environment_variables() project_variables = load_environment_variables()
""" """
@@ -31,6 +37,7 @@ DATABASE_NAMING_CONVENTION = {
"pk": "pk_%(table_name)s", "pk": "pk_%(table_name)s",
} }
class Base(DeclarativeBase): class Base(DeclarativeBase):
metadata = MetaData(naming_convention=DATABASE_NAMING_CONVENTION) metadata = MetaData(naming_convention=DATABASE_NAMING_CONVENTION)
@@ -52,18 +59,28 @@ class PowerballDraw(Base):
def __repr__(self): def __repr__(self):
return f"PowerballDraw({self.draw_date=}, {self.main_ball1=}, {self.main_ball2=}, {self.main_ball3=}, " \ return (f"PowerballDraw({self.draw_date=}, {self.main_ball1=}, "
f"{self.main_ball4=}, {self.main_ball5=}, {self.powerball=}, {self.power_play=})" f"{self.main_ball2=}, {self.main_ball3=}, ") \
f"{self.main_ball4=}, {self.main_ball5=}, {self.power
ball=}, {self.power_play=})"
def __str__(self): def __str__(self):
return f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, {self.main_ball3}, {self.main_ball4}, " \ return (f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, "
f"{self.main_ball5}], {self.powerball}, {self.power_play if self.power_play else 1}x" f"{self.main_ball3}, {self.main_ball4}, ") \
f"{self.main_ball5}], {self.powe
rball}, {self.power_play if self.power_play else 1}x"
def __eq__(self, other): def __eq__(self, other):
return self.draw_date == other.draw_date and self.main_ball1 == other.main_ball1 and \ return (self.draw_date == other.draw_date and self.main_ball1 ==
self.main_ball2 == other.main_ball2 and self.main_ball3 == other.main_ball3 and \ other.main_ball1 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 == other.main_ball5 and \ self.main_ball2 == other.main_ball2 and self.main_ball3 ==
self.powerball == other.powerball and self.power_play == other.power_play other.main_ball3 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 ==
other.main_ball5 and \
self.powerball == other.powerball and self.power_play ==
other.power_play)
class MegaMillionsDraw(Base): class MegaMillionsDraw(Base):
@@ -80,18 +97,28 @@ class MegaMillionsDraw(Base):
def __repr__(self): def __repr__(self):
return f"MegaMillionsDraw({self.draw_date=}, {self.main_ball1=}, {self.main_ball2=}, {self.main_ball3=}, " \ return (f"MegaMillionsDraw({self.draw_date=}, {self.main_ball1=}, "
f"{self.main_ball4=}, {self.main_ball5=}, {self.mega_ball=}, {self.megaplier=})" f"{self.main_ball2=}, {self.main_ball3=}, ") \
f"{self.main_ball4=}, {self.main_ball5=}, {self.mega_
ball=}, {self.megaplier=})"
def __str__(self): def __str__(self):
return f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, {self.main_ball3}, {self.main_ball4}, " \ return (f"{self.draw_date}, [{self.main_ball1}, {self.main_ball2}, "
f"{self.main_ball5}], {self.mega_ball}, {self.megaplier if self.megaplier else 1}x" f"{self.main_ball3}, {self.main_ball4}, ") \
f"{self.main_ball5}], {self.mega
_ball}, {self.megaplier if self.megaplier else 1}x"
def __eq__(self, other): def __eq__(self, other):
return self.draw_date == other.draw_date and self.main_ball1 == other.main_ball1 and \ return (self.draw_date == other.draw_date and self.main_ball1 ==
self.main_ball2 == other.main_ball2 and self.main_ball3 == other.main_ball3 and \ other.main_ball1 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 == other.main_ball5 and \ self.main_ball2 == other.main_ball2 and self.main_ball3 ==
self.mega_ball == other.mega_ball and self.megaplier == other.megaplier other.main_ball3 and \
self.main_ball4 == other.main_ball4 and self.main_ball5 ==
other.main_ball5 and \
self.mega_ball == other.mega_ball and self.megaplier ==
other.megaplier)
class DataAccessLayer: class DataAccessLayer:
@@ -112,13 +139,18 @@ class DataAccessLayer:
class DataAccessLayer2: class DataAccessLayer2:
def __init__(self): def __init__(self):
self.engine = create_engine(project_variables["LP_DATABASE_URL"], echo=False) self.engine = create_engine(
project_variables["LP_DATABASE_URL"], echo=False,
)
if not database_exists(self.engine.url): if not database_exists(self.engine.url):
create_database(self.engine.url) create_database(self.engine.url)
Base.metadata.create_all(self.engine) Base.metadata.create_all(self.engine)
self.session_local: sessionmaker[Session] = sessionmaker(bind=self.engine) self.session_local: sessionmaker[Session] = sessionmaker(
bind=self.engine,
)
self.session_factory = self.session_local self.session_factory = self.session_local
def execute_query(self, query) -> Sequence[Row[Any]]: def execute_query(self, query) -> Sequence[Row[Any]]:
with self.session_factory() as session: with self.session_factory() as session:
try: try:
@@ -127,6 +159,7 @@ class DataAccessLayer2:
finally: finally:
session.close() session.close()
def add(self, record: Table) -> None: def add(self, record: Table) -> None:
with self.session_factory() as session: with self.session_factory() as session:
try: try:
@@ -139,13 +172,17 @@ class DataAccessLayer2:
finally: finally:
session.close() session.close()
def count(self, table_name: str, from_date: date | None = None) -> int: def count(self, table_name: str, from_date: date | None = None) -> int:
with self.session_factory() as session: with self.session_factory() as session:
if from_date: if from_date:
return session.query(table_name).where(table_name.draw_date >= from_date).count() return session.query(table_name).where(
table_name.draw_date >= from_date,
).count()
else: else:
return session.query(table_name).count() return session.query(table_name).count()
def most_recent(self, table_name: Table) -> Row[Table] | None: def most_recent(self, table_name: Table) -> Row[Table] | None:
with self.session_factory() as session: with self.session_factory() as session:
return session.query(func.max(table_name.draw_date)).first() return session.query(func.max(table_name.draw_date)).first()