Coverage for src/lexigram/admin/sql_dialect.py: 100%

1"""SQL dialect helpers for direct-SQL stores. 

2 

3The admin-security stores issue raw DDL/DML that uses Postgres-only 

4syntax (``gen_random_uuid``, ``TIMESTAMPTZ``, ``NOW()``, ``INTERVAL``). 

5These helpers centralise the dialect branch so stores keep a single 

6code path that works on both Postgres and SQLite. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, Final 

12 

13from lexigram.contracts.data.sql.sql_dialect import SQLDialect 

14 

15if TYPE_CHECKING: 

16 from lexigram.contracts.data import DatabaseProviderProtocol 

17 

18 

19# Postgres drivers report ``database_type`` as ``postgres``, the enum 

20# name is ``POSTGRESQL`` — map the driver identifier to the enum member. 

21_DIALECT_ALIASES: Final[dict[str, SQLDialect]] = { 

22 "postgres": SQLDialect.POSTGRESQL, 

23} 

24 

25 

26def _dialect(db: DatabaseProviderProtocol) -> str: 

27 raw = (getattr(db, "database_type", "") or "").lower() 

28 if raw in _DIALECT_ALIASES: 

29 return _DIALECT_ALIASES[raw] 

30 for dialect in SQLDialect: 

31 if raw == dialect.value or raw == dialect.name.lower(): 

32 return dialect 

33 return raw 

34 

35 

36def is_postgres(db: DatabaseProviderProtocol) -> bool: 

37 """Return ``True`` when the provider targets Postgres. 

38 

39 Args: 

40 db: The database provider. 

41 

42 Returns: 

43 ``True`` for Postgres/PostgreSQL providers; ``False`` otherwise. 

44 """ 

45 return _dialect(db) == SQLDialect.POSTGRESQL 

46 

47 

48def now_expr(db: DatabaseProviderProtocol) -> str: 

49 """Return the dialect's current-timestamp expression. 

50 

51 Args: 

52 db: The database provider. 

53 

54 Returns: 

55 ``NOW()`` on Postgres, ``CURRENT_TIMESTAMP`` elsewhere. 

56 """ 

57 return "NOW()" if is_postgres(db) else "CURRENT_TIMESTAMP" 

58 

59 

60def since_expr(db: DatabaseProviderProtocol, seconds: int) -> str: 

61 """Return the dialect's "now minus N seconds" expression. 

62 

63 Args: 

64 db: The database provider. 

65 seconds: Look-back window in seconds. 

66 

67 Returns: 

68 Postgres ``NOW() - INTERVAL 'N seconds'`` or SQLite 

69 ``datetime('now', '-N seconds')``. 

70 """ 

71 if is_postgres(db): 

72 return f"NOW() - INTERVAL '{int(seconds)} seconds'" 

73 return f"datetime('now', '-{int(seconds)} seconds')"