41 lines
1.4 KiB
Python
41 lines
1.4 KiB
Python
"""
|
|
collect.py
|
|
|
|
Functions for collecting data for insertion into the databse or pulling data from the database for analysis
|
|
"""
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from time import sleep
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from data.database import Base, dal, MegaMillionsDraw, PowerballDraw, project_variables
|
|
import util.drawing as drawutil
|
|
import util.environment as envutil
|
|
import util.scrape as scrapeutil
|
|
|
|
project_variables = envutil.load_environment_variables()
|
|
|
|
|
|
def get_record_count(table_name: str) -> str | None:
|
|
"""get the most recent draw record from the specified table"""
|
|
if drawutil.check_table_name(table_name):
|
|
dal.connect()
|
|
session = dal.Session()
|
|
target_table = Base.registry._class_registry.get(table_name)
|
|
return session.query(func.count(target_table.draw_date)).first()[0]
|
|
else:
|
|
raise ValueError('An invalid table name was specified.')
|
|
|
|
|
|
def get_recent_draw_date(table_name: str) -> datetime | None:
|
|
"""get the most recent draw date from the specified table"""
|
|
if drawutil.check_table_name(table_name):
|
|
dal.connect()
|
|
session = dal.Session()
|
|
target_table = Base.registry._class_registry.get(table_name)
|
|
return session.query(func.max(target_table.draw_date)).first()[0]
|
|
else:
|
|
raise ValueError('An invalid table name was specified.')
|