didier/database/crud/wordle.py

79 lines
2.3 KiB
Python
Raw Normal View History

2022-08-29 20:24:42 +02:00
import datetime
2022-07-25 22:58:02 +02:00
from typing import Optional
2022-08-29 20:24:42 +02:00
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
2022-08-29 20:49:29 +02:00
from database.crud.users import get_or_add_user
2022-08-29 20:24:42 +02:00
from database.schemas import WordleGuess, WordleWord
2022-07-25 22:58:02 +02:00
2022-07-27 21:10:43 +02:00
__all__ = [
"get_active_wordle_game",
"make_wordle_guess",
"set_daily_word",
"reset_wordle_games",
]
2022-07-25 22:58:02 +02:00
2022-08-29 20:24:42 +02:00
async def get_active_wordle_game(session: AsyncSession, user_id: int) -> list[WordleGuess]:
2022-07-25 22:58:02 +02:00
"""Find a player's active game"""
2022-08-29 20:49:29 +02:00
await get_or_add_user(session, user_id)
2022-08-29 20:24:42 +02:00
statement = select(WordleGuess).where(WordleGuess.user_id == user_id)
guesses = (await session.execute(statement)).scalars().all()
return guesses
2022-07-25 22:58:02 +02:00
2022-08-29 20:24:42 +02:00
async def make_wordle_guess(session: AsyncSession, user_id: int, guess: str):
2022-07-25 22:58:02 +02:00
"""Make a guess in your current game"""
2022-08-29 20:24:42 +02:00
guess_instance = WordleGuess(user_id=user_id, guess=guess)
session.add(guess_instance)
await session.commit()
2022-08-29 20:24:42 +02:00
async def get_daily_word(session: AsyncSession) -> Optional[WordleWord]:
"""Get the word of today"""
2022-08-29 20:24:42 +02:00
statement = select(WordleWord).where(WordleWord.day == datetime.date.today())
row = (await session.execute(statement)).scalar_one_or_none()
2022-07-27 21:10:43 +02:00
2022-08-29 20:24:42 +02:00
if row is None:
return None
2022-08-29 20:24:42 +02:00
return row
2022-08-29 20:24:42 +02:00
async def set_daily_word(session: AsyncSession, word: str, *, forced: bool = False) -> str:
"""Set the word of today
This does NOT overwrite the existing word if there is one, so that it can safely run
2022-07-27 21:10:43 +02:00
on startup every time.
In order to always overwrite the current word, set the "forced"-kwarg to True.
Returns the word that was chosen. If one already existed, return that instead.
"""
2022-08-29 20:24:42 +02:00
current_word = await get_daily_word(session)
2022-07-27 21:10:43 +02:00
2022-08-29 20:24:42 +02:00
if current_word is None:
current_word = WordleWord(word=word, day=datetime.date.today())
session.add(current_word)
await session.commit()
2022-08-29 20:24:42 +02:00
# Remove all active games
await reset_wordle_games(session)
elif forced:
current_word.word = word
current_word.day = datetime.date.today()
session.add(current_word)
await session.commit()
2022-07-27 21:10:43 +02:00
2022-08-29 20:24:42 +02:00
# Remove all active games
await reset_wordle_games(session)
2022-07-27 21:10:43 +02:00
2022-08-29 20:24:42 +02:00
return current_word.word
2022-07-27 21:10:43 +02:00
2022-08-29 20:24:42 +02:00
async def reset_wordle_games(session: AsyncSession):
2022-07-27 21:10:43 +02:00
"""Reset all active games"""
2022-08-29 20:24:42 +02:00
statement = delete(WordleGuess)
await session.execute(statement)