From 355986a59fdb90ca0d854d41019925819123f1da Mon Sep 17 00:00:00 2001 From: Claude Code Date: Wed, 1 Apr 2026 16:06:00 -0500 Subject: [PATCH] fix: replace deprecated datetime.utcnow() with timezone-aware alternative Co-Authored-By: Claude Sonnet 4.6 --- app/curator.py | 6 +++--- app/proxy_handler.py | 6 +++--- app/qdrant_service.py | 6 +++--- app/utils.py | 4 ++-- tests/test_curator.py | 8 ++++---- tests/test_utils.py | 12 ++++++------ 6 files changed, 21 insertions(+), 21 deletions(-) diff --git a/app/curator.py b/app/curator.py index f6685ea..58cc0ab 100644 --- a/app/curator.py +++ b/app/curator.py @@ -6,7 +6,7 @@ The prompt determines behavior based on current date. """ import logging import os -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from typing import List, Dict, Any, Optional from pathlib import Path import httpx @@ -49,7 +49,7 @@ class Curator: Otherwise runs daily mode (processes recent 24h only). The prompt determines behavior based on current date. """ - current_date = datetime.utcnow() + current_date = datetime.now(timezone.utc) is_monthly = current_date.day == 1 mode = "MONTHLY" if is_monthly else "DAILY" @@ -169,7 +169,7 @@ Remember: Respond with ONLY valid JSON. No markdown, no explanations, just the J return True try: mem_time = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) - cutoff = datetime.utcnow() - timedelta(hours=hours) + cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=hours) return mem_time.replace(tzinfo=None) > cutoff except (ValueError, TypeError): logger.debug(f"Could not parse timestamp: {timestamp}") diff --git a/app/proxy_handler.py b/app/proxy_handler.py index 357dd7d..f9dc725 100644 --- a/app/proxy_handler.py +++ b/app/proxy_handler.py @@ -48,17 +48,17 @@ def debug_log(category: str, message: str, data: dict = None): if not config.debug: return - from datetime import datetime + from datetime import datetime, timezone # Create logs directory log_dir = DEBUG_LOG_DIR log_dir.mkdir(parents=True, exist_ok=True) - today = datetime.utcnow().strftime("%Y-%m-%d") + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") log_path = log_dir / f"debug_{today}.log" entry = { - "timestamp": datetime.utcnow().isoformat() + "Z", + "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "category": category, "message": message } diff --git a/app/qdrant_service.py b/app/qdrant_service.py index 63b9fdd..90890f4 100644 --- a/app/qdrant_service.py +++ b/app/qdrant_service.py @@ -2,7 +2,7 @@ from qdrant_client import AsyncQdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue from typing import List, Dict, Any, Optional -from datetime import datetime +from datetime import datetime, timezone import uuid import logging import httpx @@ -54,7 +54,7 @@ class QdrantService: point_id = str(uuid.uuid4()) embedding = await self.get_embedding(content) - timestamp = datetime.utcnow().isoformat() + "Z" + timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") text = content if role == "user": text = f"User: {content}" @@ -85,7 +85,7 @@ class QdrantService: """Store a complete Q&A turn as one document.""" await self._ensure_collection() - timestamp = datetime.utcnow().isoformat() + "Z" + timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") text = f"User: {user_question}\nAssistant: {assistant_answer}\nTimestamp: {timestamp}" point_id = str(uuid.uuid4()) diff --git a/app/utils.py b/app/utils.py index 532a8a5..606a668 100644 --- a/app/utils.py +++ b/app/utils.py @@ -3,7 +3,7 @@ from .config import config import tiktoken import os from typing import List, Dict, Optional -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path # Use cl100k_base encoding (GPT-4 compatible) @@ -56,7 +56,7 @@ def truncate_by_tokens(text: str, max_tokens: int) -> str: def filter_memories_by_time(memories: List[Dict], hours: int = 24) -> List[Dict]: """Filter memories from the last N hours.""" - cutoff = datetime.utcnow() - timedelta(hours=hours) + cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=hours) filtered = [] for mem in memories: ts = mem.get("timestamp") diff --git a/tests/test_curator.py b/tests/test_curator.py index 26aaa49..5e91998 100644 --- a/tests/test_curator.py +++ b/tests/test_curator.py @@ -2,7 +2,7 @@ import pytest import json import os -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import MagicMock, patch @@ -77,14 +77,14 @@ class TestIsRecent: def test_memory_within_window(self): """Memory timestamped 1 hour ago is recent (within 24h).""" curator, _ = make_curator() - ts = (datetime.utcnow() - timedelta(hours=1)).isoformat() + "Z" + ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)).isoformat() + "Z" memory = {"timestamp": ts} assert curator._is_recent(memory, hours=24) is True def test_memory_outside_window(self): """Memory timestamped 48 hours ago is not recent.""" curator, _ = make_curator() - ts = (datetime.utcnow() - timedelta(hours=48)).isoformat() + "Z" + ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=48)).isoformat() + "Z" memory = {"timestamp": ts} assert curator._is_recent(memory, hours=24) is False @@ -109,7 +109,7 @@ class TestIsRecent: def test_boundary_edge_just_inside(self): """Memory at exactly hours-1 minutes ago should be recent.""" curator, _ = make_curator() - ts = (datetime.utcnow() - timedelta(hours=23, minutes=59)).isoformat() + "Z" + ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=23, minutes=59)).isoformat() + "Z" memory = {"timestamp": ts} assert curator._is_recent(memory, hours=24) is True diff --git a/tests/test_utils.py b/tests/test_utils.py index 63833e0..05e9f2c 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -90,20 +90,20 @@ class TestFilterMemoriesByTime: def test_includes_recent_memory(self): """Memory with timestamp in the last 24h should be included.""" - from datetime import datetime, timedelta + from datetime import datetime, timedelta, timezone from app.utils import filter_memories_by_time - ts = (datetime.utcnow() - timedelta(hours=1)).isoformat() + ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)).isoformat() memories = [{"timestamp": ts, "text": "recent"}] result = filter_memories_by_time(memories, hours=24) assert len(result) == 1 def test_excludes_old_memory(self): """Memory older than cutoff should be excluded.""" - from datetime import datetime, timedelta + from datetime import datetime, timedelta, timezone from app.utils import filter_memories_by_time - ts = (datetime.utcnow() - timedelta(hours=48)).isoformat() + ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=48)).isoformat() memories = [{"timestamp": ts, "text": "old"}] result = filter_memories_by_time(memories, hours=24) assert len(result) == 0 @@ -132,10 +132,10 @@ class TestFilterMemoriesByTime: def test_z_suffix_timestamp(self): """ISO timestamp with Z suffix should be handled correctly.""" - from datetime import datetime, timedelta + from datetime import datetime, timedelta, timezone from app.utils import filter_memories_by_time - ts = (datetime.utcnow() - timedelta(hours=1)).isoformat() + "Z" + ts = (datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(hours=1)).isoformat() + "Z" memories = [{"timestamp": ts, "text": "recent with Z"}] result = filter_memories_by_time(memories, hours=24) assert len(result) == 1