Refactor timezone handling, raise if fucked on startup
This commit is contained in:
@ -1,40 +1,32 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
import os
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import dateparser
|
||||
from loguru import logger
|
||||
|
||||
from discord_reminder_bot.settings import get_timezone
|
||||
|
||||
|
||||
def parse_time(date_to_parse: str | None, timezone: str | None = None, use_dotenv: bool = True) -> datetime.datetime | None: # noqa: FBT001, FBT002
|
||||
def parse_time(date_to_parse: str | None, timezone: str | None = os.getenv("TIMEZONE")) -> datetime.datetime | None:
|
||||
"""Parse a date string into a datetime object.
|
||||
|
||||
Args:
|
||||
date_to_parse(str): The date string to parse.
|
||||
timezone(str, optional): The timezone to use. Defaults timezone from settings.
|
||||
use_dotenv(bool, optional): Whether to load environment variables from a .env file. Defaults to True
|
||||
|
||||
Returns:
|
||||
datetime.datetime: The parsed datetime object.
|
||||
"""
|
||||
logger.info(f"Parsing date: '{date_to_parse}' with timezone: '{timezone}'")
|
||||
|
||||
if not date_to_parse:
|
||||
logger.error("No date provided to parse.")
|
||||
return None
|
||||
|
||||
if not timezone:
|
||||
timezone = get_timezone(use_dotenv)
|
||||
logger.error("No timezone provided to parse date.")
|
||||
return None
|
||||
|
||||
# Check if the timezone is valid
|
||||
try:
|
||||
tz = ZoneInfo(timezone)
|
||||
except (ZoneInfoNotFoundError, ModuleNotFoundError):
|
||||
logger.error(f"Invalid timezone provided: '{timezone}'. Using {get_timezone(use_dotenv)} instead.")
|
||||
tz = ZoneInfo("UTC")
|
||||
logger.info(f"Parsing date: '{date_to_parse}' with timezone: '{timezone}'")
|
||||
|
||||
try:
|
||||
parsed_date: datetime.datetime | None = dateparser.parse(
|
||||
@ -43,10 +35,11 @@ def parse_time(date_to_parse: str | None, timezone: str | None = None, use_doten
|
||||
"PREFER_DATES_FROM": "future",
|
||||
"TIMEZONE": f"{timezone}",
|
||||
"RETURN_AS_TIMEZONE_AWARE": True,
|
||||
"RELATIVE_BASE": datetime.datetime.now(tz=tz),
|
||||
"RELATIVE_BASE": datetime.datetime.now(tz=ZoneInfo(str(timezone))),
|
||||
},
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
except (ValueError, TypeError) as e:
|
||||
logger.error(f"Failed to parse date: '{date_to_parse}' with timezone: '{timezone}'. Error: {e}")
|
||||
return None
|
||||
|
||||
logger.debug(f"Parsed date: {parsed_date} from '{date_to_parse}'")
|
||||
|
@ -1,6 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
|
||||
import pytz
|
||||
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
|
||||
@ -24,15 +25,26 @@ def get_settings(use_dotenv: bool = True) -> dict[str, str | dict[str, SQLAlchem
|
||||
load_dotenv(verbose=True)
|
||||
|
||||
sqlite_location: str = os.getenv("SQLITE_LOCATION", default="/jobs.sqlite")
|
||||
config_timezone: str = os.getenv("TIMEZONE", default="UTC")
|
||||
bot_token: str = os.getenv("BOT_TOKEN", default="")
|
||||
log_level: str = os.getenv("LOG_LEVEL", default="INFO")
|
||||
webhook_url: str = os.getenv("WEBHOOK_URL", default="")
|
||||
|
||||
bot_token: str = os.getenv("BOT_TOKEN", default="")
|
||||
if not bot_token:
|
||||
msg = "Missing bot token. Please set the BOT_TOKEN environment variable."
|
||||
raise ValueError(msg)
|
||||
|
||||
config_timezone: str | None = os.getenv("TIMEZONE")
|
||||
if not config_timezone:
|
||||
msg = "Missing timezone. Please set the TIMEZONE environment variable."
|
||||
raise ValueError(msg)
|
||||
|
||||
# Test if the timezone is valid
|
||||
try:
|
||||
ZoneInfo(config_timezone)
|
||||
except (ZoneInfoNotFoundError, ModuleNotFoundError) as e:
|
||||
msg: str = f"Invalid timezone: {config_timezone}. Error: {e}"
|
||||
raise ValueError(msg) from e
|
||||
|
||||
jobstores: dict[str, SQLAlchemyJobStore] = {"default": SQLAlchemyJobStore(url=f"sqlite://{sqlite_location}")}
|
||||
job_defaults: dict[str, bool] = {"coalesce": True}
|
||||
scheduler = AsyncIOScheduler(
|
||||
@ -128,29 +140,3 @@ def get_webhook_url(use_dotenv: bool = True) -> str: # noqa: FBT001, FBT002
|
||||
|
||||
msg = "The webhook URL is missing from the settings."
|
||||
raise KeyError(msg)
|
||||
|
||||
|
||||
def get_timezone(use_dotenv: bool = True) -> str: # noqa: FBT001, FBT002
|
||||
"""Return the timezone.
|
||||
|
||||
Args:
|
||||
use_dotenv (bool, optional): Whether to load environment variables from a .env file. Defaults to True
|
||||
|
||||
Raises:
|
||||
TypeError: If the timezone is not a string.
|
||||
KeyError: If the timezone is missing from the settings.
|
||||
|
||||
Returns:
|
||||
str: The timezone.
|
||||
"""
|
||||
settings: dict[str, str | dict[str, SQLAlchemyJobStore] | dict[str, bool] | AsyncIOScheduler] = get_settings(use_dotenv)
|
||||
|
||||
if config_timezone := settings.get("config_timezone"):
|
||||
if not isinstance(config_timezone, str):
|
||||
msg = "The timezone is not a string."
|
||||
raise TypeError(msg)
|
||||
|
||||
return config_timezone
|
||||
|
||||
msg = "The timezone is missing from the settings."
|
||||
raise KeyError(msg)
|
||||
|
Reference in New Issue
Block a user