Coverage for node / src / stigmem_node / storage / __init__.py: 67%

1"""Storage backend registry and factory. 

2 

3Usage:: 

4 

5 from stigmem_node.storage import make_backend 

6 

7 # Relies on settings.storage_backend (and related env vars) 

8 backend = make_backend(_settings=settings) 

9 

10 with backend.connection() as conn: 

11 conn.execute("SELECT ...") 

12 

13 # Explicit path always returns a plaintext SQLiteBackend (backward-compat 

14 # for CLI tools and test fixtures that create their own temp databases). 

15 backend = make_backend(db_path="/tmp/test.db") 

16""" 

17 

18from __future__ import annotations 

19 

20import os 

21from typing import TYPE_CHECKING, Any 

22 

23from .base import StorageBackend 

24from .libsql_backend import LibSQLBackend 

25from .postgres_backend import PostgresBackend 

26from .sqlite_backend import SQLiteBackend 

27 

28if TYPE_CHECKING: 

29 pass 

30 

31__all__ = ["StorageBackend", "SQLiteBackend", "LibSQLBackend", "PostgresBackend", "make_backend"] 

32 

33 

34def make_backend( 

35 db_path: str | None = None, 

36 _settings: Any | None = None, 

37) -> StorageBackend: 

38 """Return the appropriate ``StorageBackend`` instance. 

39 

40 Args: 

41 db_path: When provided, always returns a plaintext ``SQLiteBackend`` 

42 at this path. This preserves backward-compatibility for CLI tools, 

43 ``apply_migrations(db_path=…)`` calls, and test fixtures that 

44 supply a temporary database path directly. Encryption settings 

45 are ignored when *db_path* is given. 

46 _settings: A ``Settings`` object to read ``storage_backend``, 

47 ``db_path``, ``libsql_url``, ``libsql_auth_token``, and 

48 ``at_rest_encryption`` from. Defaults to the live 

49 ``stigmem_node.settings.settings`` singleton so that test fixtures 

50 can control backend selection by patching ``db_mod.settings``. 

51 

52 Raises: 

53 RuntimeError: If ``at_rest_encryption="on"`` is set but no key source 

54 is configured (``at_rest_key_passphrase_env`` / ``at_rest_key_kms_uri`` 

55 both empty). The node refuses to start in this state. 

56 """ 

57 # An explicit path always means plaintext SQLite — backward-compat guarantee. 

58 if db_path is not None: 

59 return SQLiteBackend(db_path) 

60 

61 if _settings is None: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 from stigmem_node.settings import settings # lazy to avoid import cycles 

63 

64 _settings = settings 

65 

66 # Resolve encryption key (Phase 8 / ACM-184). load_key() raises immediately 

67 # if encryption is on but no key source is configured — this surfaces the 

68 # misconfiguration at app startup (apply_migrations call) rather than lazily. 

69 encryption_key: bytes | None = None 

70 if getattr(_settings, "at_rest_encryption", "off") == "on": 

71 from .encryption import load_key 

72 

73 encryption_key = load_key(_settings) 

74 

75 backend_name: str = getattr(_settings, "storage_backend", "sqlite") 

76 path: str = _settings.db_path 

77 

78 if backend_name == "libsql": 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 return LibSQLBackend( 

80 db_path=path, 

81 sync_url=getattr(_settings, "libsql_url", ""), 

82 auth_token=getattr(_settings, "libsql_auth_token", ""), 

83 encryption_key=encryption_key, 

84 ) 

85 

86 if backend_name == "postgres": 86 ↛ 88line 86 didn't jump to line 88 because the condition on line 86 was never true

87 # Accept STIGMEM_PG_DSN, STIGMEM_DATABASE_URL, or bare DATABASE_URL. 

88 dsn: str = ( 

89 getattr(_settings, "pg_dsn", "") 

90 or getattr(_settings, "database_url", "") 

91 or os.environ.get("DATABASE_URL", "") 

92 ) 

93 if not dsn: 

94 raise RuntimeError( 

95 "storage_backend='postgres' requires a connection string. " 

96 "Set STIGMEM_PG_DSN, STIGMEM_DATABASE_URL, or DATABASE_URL." 

97 ) 

98 schema: str = getattr(_settings, "pg_schema", "public") or "public" 

99 embed_enabled_pg: bool = getattr(_settings, "embed_enabled", False) 

100 embed_dimension_pg: int = int(getattr(_settings, "embed_dimension", 768)) 

101 return PostgresBackend( 

102 dsn=dsn, 

103 schema=schema, 

104 pool_min=int(getattr(_settings, "postgres_pool_min", 2)), 

105 pool_max=int(getattr(_settings, "postgres_pool_max", 10)), 

106 embed_enabled=embed_enabled_pg, 

107 embed_dimension=embed_dimension_pg, 

108 ) 

109 

110 embed_enabled: bool = getattr(_settings, "embed_enabled", False) 

111 embed_dimension: int = int(getattr(_settings, "embed_dimension", 768)) 

112 return SQLiteBackend( 

113 path, 

114 encryption_key=encryption_key, 

115 embed_enabled=embed_enabled, 

116 embed_dimension=embed_dimension, 

117 )