65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
"""Партиции аналитики по дням.
|
||
|
||
Создаются заранее (вчера/сегодня/два дня вперёд), старые **не удаляются**:
|
||
история событий и дневных счётчиков хранится бессрочно.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from datetime import UTC, date, datetime, timedelta
|
||
|
||
from common.config import Settings
|
||
from common.db.pool import get_pool
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
#: Сколько дней вперёд создавать партиции.
|
||
DAYS_AHEAD = 2
|
||
#: Сколько дней назад проверять (на случай простоя).
|
||
DAYS_BACK = 1
|
||
#: Интервал фоновой проверки партиций.
|
||
PARTITION_INTERVAL = 6 * 60 * 60
|
||
|
||
|
||
def _event_partition(day: date) -> str:
|
||
start = f"{day.isoformat()} 00:00:00+00"
|
||
end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00+00"
|
||
return (
|
||
f"CREATE TABLE IF NOT EXISTS analytics.events_{day:%Y%m%d} "
|
||
f"PARTITION OF analytics.events FOR VALUES FROM ('{start}') TO ('{end}')"
|
||
)
|
||
|
||
|
||
def _messages_partition(day: date) -> str:
|
||
start = day.isoformat()
|
||
end = (day + timedelta(days=1)).isoformat()
|
||
return (
|
||
f"CREATE TABLE IF NOT EXISTS analytics.messages_daily_{day:%Y%m%d} "
|
||
f"PARTITION OF analytics.messages_daily FOR VALUES FROM ('{start}') TO ('{end}')"
|
||
)
|
||
|
||
|
||
async def ensure_partitions(settings: Settings) -> None:
|
||
"""Создать недостающие партиции на несколько дней вперёд."""
|
||
utc_today = datetime.now(UTC).date()
|
||
local_today = datetime.now(settings.tzinfo).date()
|
||
|
||
async with get_pool().connection() as conn:
|
||
for offset in range(-DAYS_BACK, DAYS_AHEAD + 1):
|
||
await conn.execute(_event_partition(utc_today + timedelta(days=offset)))
|
||
await conn.execute(_messages_partition(local_today + timedelta(days=offset)))
|
||
|
||
|
||
async def run_partitions(settings: Settings) -> None:
|
||
"""Периодически проверять партиции; ошибки не роняют бота."""
|
||
while True:
|
||
try:
|
||
await ensure_partitions(settings)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
logger.warning("Не удалось создать партиции аналитики", exc_info=True)
|
||
await asyncio.sleep(PARTITION_INTERVAL)
|