fix(core): add utcnow()/utcnow_iso() helpers, migrate core.py usages

Replace deprecated datetime.utcnow() with datetime.now(UTC).
- utcnow() -> datetime: for in-memory datetime math
- utcnow_iso() -> str: strftime format preserving existing SQLite TEXT format

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Deeman
2026-02-24 10:12:50 +01:00
parent a35036807e
commit f76d2889e5

View File

@@ -10,7 +10,7 @@ import re
import secrets import secrets
import unicodedata import unicodedata
from contextvars import ContextVar from contextvars import ContextVar
from datetime import datetime, timedelta from datetime import UTC, datetime, timedelta
from functools import wraps from functools import wraps
from pathlib import Path from pathlib import Path
@@ -88,6 +88,26 @@ class Config:
config = Config() config = Config()
# =============================================================================
# Datetime helpers
# =============================================================================
def utcnow() -> datetime:
"""Timezone-aware UTC now (replaces deprecated datetime.utcnow())."""
return datetime.now(UTC)
def utcnow_iso() -> str:
"""UTC now as naive ISO string for SQLite TEXT columns.
Produces YYYY-MM-DDTHH:MM:SS (no +00:00 suffix) to match the existing
format stored in the DB so lexicographic SQL comparisons keep working.
"""
return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
# ============================================================================= # =============================================================================
# Database # Database
# ============================================================================= # =============================================================================
@@ -528,17 +548,18 @@ async def check_rate_limit(key: str, limit: int = None, window: int = None) -> t
""" """
limit = limit or config.RATE_LIMIT_REQUESTS limit = limit or config.RATE_LIMIT_REQUESTS
window = window or config.RATE_LIMIT_WINDOW window = window or config.RATE_LIMIT_WINDOW
now = datetime.utcnow() now = utcnow()
window_start = now - timedelta(seconds=window) window_start = now - timedelta(seconds=window)
# Clean old entries and count recent # Clean old entries and count recent
await execute( await execute(
"DELETE FROM rate_limits WHERE key = ? AND timestamp < ?", (key, window_start.isoformat()) "DELETE FROM rate_limits WHERE key = ? AND timestamp < ?",
(key, window_start.strftime("%Y-%m-%dT%H:%M:%S")),
) )
result = await fetch_one( result = await fetch_one(
"SELECT COUNT(*) as count FROM rate_limits WHERE key = ? AND timestamp > ?", "SELECT COUNT(*) as count FROM rate_limits WHERE key = ? AND timestamp > ?",
(key, window_start.isoformat()), (key, window_start.strftime("%Y-%m-%dT%H:%M:%S")),
) )
count = result["count"] if result else 0 count = result["count"] if result else 0
@@ -552,7 +573,10 @@ async def check_rate_limit(key: str, limit: int = None, window: int = None) -> t
return False, info return False, info
# Record this request # Record this request
await execute("INSERT INTO rate_limits (key, timestamp) VALUES (?, ?)", (key, now.isoformat())) await execute(
"INSERT INTO rate_limits (key, timestamp) VALUES (?, ?)",
(key, now.strftime("%Y-%m-%dT%H:%M:%S")),
)
return True, info return True, info
@@ -628,7 +652,7 @@ async def soft_delete(table: str, id: int) -> bool:
"""Mark record as deleted.""" """Mark record as deleted."""
result = await execute( result = await execute(
f"UPDATE {table} SET deleted_at = ? WHERE id = ? AND deleted_at IS NULL", f"UPDATE {table} SET deleted_at = ? WHERE id = ? AND deleted_at IS NULL",
(datetime.utcnow().isoformat(), id), (utcnow_iso(), id),
) )
return result > 0 return result > 0
@@ -647,7 +671,7 @@ async def hard_delete(table: str, id: int) -> bool:
async def purge_deleted(table: str, days: int = 30) -> int: async def purge_deleted(table: str, days: int = 30) -> int:
"""Purge records deleted more than X days ago.""" """Purge records deleted more than X days ago."""
cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat() cutoff = (utcnow() - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%S")
return await execute( return await execute(
f"DELETE FROM {table} WHERE deleted_at IS NOT NULL AND deleted_at < ?", (cutoff,) f"DELETE FROM {table} WHERE deleted_at IS NOT NULL AND deleted_at < ?", (cutoff,)
) )