Coverage for src/dataknobs_data/factory.py: 21%

1"""Backend factory for dynamic database creation.""" 

2 

3import logging 

4from typing import Any, Dict, Optional 

5 

6from dataknobs_config import FactoryBase 

7from dataknobs_data.database import SyncDatabase 

8 

9logger = logging.getLogger(__name__) 

10 

11 

12class DatabaseFactory(FactoryBase): 

13 """Factory for creating database backends dynamically. 

14  

15 This factory allows creating different database implementations 

16 based on configuration, supporting all available backends. 

17  

18 Configuration Options: 

19 backend (str): Backend type (memory, file, postgres, elasticsearch, s3) 

20 **kwargs: Backend-specific configuration options 

21  

22 Example Configuration: 

23 databases: 

24 - name: main 

25 factory: database 

26 backend: postgres 

27 host: localhost 

28 database: myapp 

29  

30 - name: cache 

31 factory: database 

32 backend: memory 

33  

34 - name: archive 

35 factory: database 

36 backend: s3 

37 bucket: my-archive-bucket 

38 prefix: archives/ 

39 """ 

40 

41 def create(self, **config) -> SyncDatabase: 

42 """Create a database instance based on configuration. 

43  

44 Args: 

45 **config: Configuration including 'backend' field and backend-specific options 

46  

47 Returns: 

48 Instance of appropriate database backend 

49  

50 Raises: 

51 ValueError: If backend type is not recognized or not available 

52 """ 

53 backend_type = config.pop("backend", "memory").lower() 

54 

55 logger.info(f"Creating database with backend: {backend_type}") 

56 

57 if backend_type in ("memory", "mem"): 

58 from dataknobs_data.backends.memory import SyncMemoryDatabase 

59 return SyncMemoryDatabase.from_config(config) 

60 

61 elif backend_type == "file": 

62 from dataknobs_data.backends.file import SyncFileDatabase 

63 return SyncFileDatabase.from_config(config) 

64 

65 elif backend_type in ("postgres", "postgresql", "pg"): 

66 try: 

67 from dataknobs_data.backends.postgres import SyncPostgresDatabase 

68 return SyncPostgresDatabase.from_config(config) 

69 except ImportError as e: 

70 raise ValueError( 

71 f"PostgreSQL backend requires psycopg2. " 

72 f"Install with: pip install dataknobs-data[postgres]" 

73 ) from e 

74 

75 elif backend_type in ("elasticsearch", "es"): 

76 try: 

77 from dataknobs_data.backends.elasticsearch import SyncElasticsearchDatabase 

78 return SyncElasticsearchDatabase.from_config(config) 

79 except ImportError as e: 

80 raise ValueError( 

81 f"Elasticsearch backend requires elasticsearch package. " 

82 f"Install with: pip install dataknobs-data[elasticsearch]" 

83 ) from e 

84 

85 elif backend_type == "s3": 

86 try: 

87 from dataknobs_data.backends.s3 import SyncS3Database 

88 return SyncS3Database.from_config(config) 

89 except ImportError as e: 

90 raise ValueError( 

91 f"S3 backend requires boto3. " 

92 f"Install with: pip install dataknobs-data[s3]" 

93 ) from e 

94 

95 else: 

96 raise ValueError( 

97 f"Unknown backend type: {backend_type}. " 

98 f"Available backends: memory, file, postgres, elasticsearch, s3" 

99 ) 

100 

101 def get_backend_info(self, backend_type: str) -> Dict[str, Any]: 

102 """Get information about a specific backend. 

103  

104 Args: 

105 backend_type: Name of the backend 

106  

107 Returns: 

108 Dictionary with backend information 

109 """ 

110 info = { 

111 "memory": { 

112 "description": "In-memory storage for testing and caching", 

113 "persistent": False, 

114 "requires_install": False, 

115 "config_options": { 

116 "initial_data": "Optional initial data dictionary" 

117 } 

118 }, 

119 "file": { 

120 "description": "File-based storage (JSON, CSV, Parquet)", 

121 "persistent": True, 

122 "requires_install": False, 

123 "config_options": { 

124 "path": "Path to the file (required)", 

125 "format": "File format: json, csv, parquet (default: json)", 

126 "compression": "Optional compression: gzip, bz2, xz" 

127 } 

128 }, 

129 "postgres": { 

130 "description": "PostgreSQL database backend", 

131 "persistent": True, 

132 "requires_install": "pip install dataknobs-data[postgres]", 

133 "config_options": { 

134 "host": "Database host (required)", 

135 "port": "Database port (default: 5432)", 

136 "database": "Database name (required)", 

137 "user": "Username (required)", 

138 "password": "Password (required)", 

139 "table": "Table name (default: records)" 

140 } 

141 }, 

142 "elasticsearch": { 

143 "description": "Elasticsearch search engine backend", 

144 "persistent": True, 

145 "requires_install": "pip install dataknobs-data[elasticsearch]", 

146 "config_options": { 

147 "hosts": "List of host URLs (required)", 

148 "index": "Index name (required)", 

149 "doc_type": "Document type (default: _doc)", 

150 "username": "Optional username", 

151 "password": "Optional password" 

152 } 

153 }, 

154 "s3": { 

155 "description": "AWS S3 object storage backend", 

156 "persistent": True, 

157 "requires_install": "pip install dataknobs-data[s3]", 

158 "config_options": { 

159 "bucket": "S3 bucket name (required)", 

160 "prefix": "Object key prefix (default: records/)", 

161 "region": "AWS region (default: us-east-1)", 

162 "endpoint_url": "Custom endpoint for S3-compatible services", 

163 "access_key_id": "AWS access key (or use IAM role)", 

164 "secret_access_key": "AWS secret key (or use IAM role)" 

165 } 

166 } 

167 } 

168 

169 return info.get(backend_type.lower(), { 

170 "description": "Unknown backend", 

171 "error": f"Backend '{backend_type}' not recognized" 

172 }) 

173 

174 

175class AsyncDatabaseFactory(FactoryBase): 

176 """Factory for creating async database backends. 

177  

178 Note: Currently only some backends support async operations. 

179 """ 

180 

181 def create(self, **config) -> Any: 

182 """Create an async database instance. 

183  

184 Args: 

185 **config: Configuration including 'backend' field 

186  

187 Returns: 

188 Instance of appropriate async database backend 

189  

190 Raises: 

191 ValueError: If backend doesn't support async operations 

192 """ 

193 backend_type = config.pop("backend", "memory").lower() 

194 

195 if backend_type in ("memory", "mem"): 

196 from dataknobs_data.backends.memory import AsyncMemoryDatabase 

197 return AsyncMemoryDatabase.from_config(config) 

198 

199 elif backend_type == "file": 

200 from dataknobs_data.backends.file import AsyncFileDatabase 

201 return AsyncFileDatabase.from_config(config) 

202 

203 elif backend_type in ("postgres", "postgresql", "pg"): 

204 from dataknobs_data.backends.postgres import AsyncPostgresDatabase 

205 return AsyncPostgresDatabase.from_config(config) 

206 

207 elif backend_type in ("elasticsearch", "es"): 

208 from dataknobs_data.backends.elasticsearch import AsyncElasticsearchDatabase 

209 return AsyncElasticsearchDatabase.from_config(config) 

210 

211 elif backend_type == "s3": 

212 from dataknobs_data.backends.s3 import AsyncS3Database 

213 return AsyncS3Database.from_config(config) 

214 

215 else: 

216 raise ValueError( 

217 f"Backend '{backend_type}' does not support async operations yet. " 

218 f"Available async backends: memory, file, postgres, elasticsearch, s3" 

219 ) 

220 

221 

222# Create singleton instances for registration 

223database_factory = DatabaseFactory() 

224async_database_factory = AsyncDatabaseFactory()