28 lines
700 B
Python
28 lines
700 B
Python
from datetime import datetime
|
|
|
|
import asyncpg
|
|
|
|
DEFAULT_TZ = "UTC"
|
|
MAX_DAYS = 365
|
|
|
|
_timezones: set[str] = set()
|
|
|
|
|
|
async def known_timezone(pool: asyncpg.Pool, tz: str) -> bool:
|
|
if not _timezones:
|
|
rows = await pool.fetch("SELECT name FROM pg_timezone_names")
|
|
_timezones.update(row["name"] for row in rows)
|
|
return tz in _timezones
|
|
|
|
|
|
async def period_bounds(
|
|
pool: asyncpg.Pool, days: int, tz: str
|
|
) -> tuple[datetime, datetime]:
|
|
row = await pool.fetchrow(
|
|
"SELECT (date_trunc('day', now() AT TIME ZONE $1) "
|
|
"- make_interval(days => $2 - 1)) AT TIME ZONE $1 AS since, now() AS until",
|
|
tz,
|
|
days,
|
|
)
|
|
return row["since"], row["until"]
|