Coverage for src / lexigram / contracts / data / sql / query_log.py: 0%

1"""Database query logging contracts (renamed from ``logging.py``). 

2 

3This module contains the contracts for database query logging. It was 

4created during the 2026 reorganization when the previous 

5``data/logging.py`` module was renamed; the old path is no longer supported. 

6Consumers should import directly from ``lexigram.contracts.data``. 

7""" 

8 

9from __future__ import annotations 

10 

11from dataclasses import dataclass, field 

12from datetime import datetime 

13from typing import Any, Protocol, runtime_checkable 

14 

15 

16@dataclass(frozen=True) 

17class QueryLogEntry: 

18 """Represents a database query log entry. 

19 

20 This is a concrete dataclass rather than a Protocol so that 

21 callers can instantiate it directly. 

22 """ 

23 

24 sql: str 

25 params: tuple[Any, ...] | None = None 

26 execution_time: float = 0.0 

27 timestamp: datetime = field(default_factory=datetime.now) 

28 success: bool = True 

29 error_message: str | None = None 

30 connection_id: str | None = None 

31 transaction_id: str | None = None 

32 user_id: str | None = None 

33 request_id: str | None = None 

34 trace_id: str | None = None 

35 

36 

37@runtime_checkable 

38class QueryLoggerProtocol(Protocol): 

39 """Protocol for logging database queries. 

40 

41 Implement this protocol to enable query logging in database providers. 

42 The concrete :class:`lexigram.sql.logging.BaseQueryLogger` implements 

43 this interface and adds helper methods for querying the stored entries. 

44 """ 

45 

46 async def log_query( 

47 self, 

48 entry: QueryLogEntry, 

49 ) -> None: # pragma: no cover - protocol 

50 """Log a query execution.""" 

51 ... 

52 

53 async def get_recent_queries(self, limit: int = 100) -> list[QueryLogEntry]: 

54 """Return the most recent ``limit`` entries.""" 

55 ... 

56 

57 async def get_slow_queries( 

58 self, 

59 threshold_seconds: float, 

60 limit: int = 50, 

61 ) -> list[QueryLogEntry]: 

62 """Return entries whose execution time exceeds ``threshold_seconds``.""" 

63 ... 

64 

65 async def get_query_stats(self, time_range_seconds: int = 3600) -> dict[str, Any]: 

66 """Return aggregated statistics over a time window.""" 

67 ... 

68 

69 

70__all__ = [ 

71 "QueryLogEntry", 

72 "QueryLoggerProtocol", 

73]