Coverage for src / mcp_server_langgraph / compliance / gdpr / factory.py: 74%
43 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 00:43 +0000
1"""
2GDPR Storage Factory
4Creates storage backend instances based on configuration.
5Supports multiple backends:
6- PostgreSQL (production-ready, recommended)
7- In-memory (development/testing only)
9Pattern: Factory pattern with dependency injection
10"""
12from typing import Any, Literal
14from mcp_server_langgraph.compliance.gdpr.postgres_storage import (
15 PostgresAuditLogStore,
16 PostgresConsentStore,
17 PostgresConversationStore,
18 PostgresPreferencesStore,
19 PostgresUserProfileStore,
20)
21from mcp_server_langgraph.compliance.gdpr.storage import (
22 AuditLogStore,
23 ConsentStore,
24 ConversationStore,
25 InMemoryAuditLogStore,
26 InMemoryConsentStore,
27 InMemoryConversationStore,
28 InMemoryPreferencesStore,
29 InMemoryUserProfileStore,
30 PreferencesStore,
31 UserProfileStore,
32)
35class GDPRStorage:
36 """
37 Container for all GDPR storage backends
39 Provides unified access to all storage components:
40 - user_profiles: User profile storage
41 - preferences: User preferences storage
42 - consents: Consent record storage
43 - conversations: Conversation storage
44 - audit_logs: Audit log storage
45 """
47 def __init__(
48 self,
49 user_profiles: UserProfileStore,
50 preferences: PreferencesStore,
51 consents: ConsentStore,
52 conversations: ConversationStore,
53 audit_logs: AuditLogStore,
54 ):
55 self.user_profiles = user_profiles
56 self.preferences = preferences
57 self.consents = consents
58 self.conversations = conversations
59 self.audit_logs = audit_logs
62async def create_postgres_storage(postgres_url: str) -> GDPRStorage:
63 """
64 Create PostgreSQL-backed GDPR storage (RECOMMENDED for production)
66 Uses retry logic with exponential backoff to handle transient connection failures.
68 Args:
69 postgres_url: PostgreSQL connection URL
70 Example: postgresql://user:pass@localhost:5432/gdpr
72 Returns:
73 GDPRStorage instance with PostgreSQL backends
75 Raises:
76 asyncpg.PostgresError: If connection fails after retries
77 Exception: If retries are exhausted
79 Example:
80 >>> storage = await create_postgres_storage("postgresql://postgres:postgres@localhost:5432/gdpr")
81 >>> profile = await storage.user_profiles.get("user:alice")
82 """
83 from mcp_server_langgraph.infrastructure.database import create_connection_pool
85 # Create connection pool with retry logic
86 # Retries: 3, Backoff: 1s, 2s, 4s (with jitter)
87 pool = await create_connection_pool(
88 postgres_url,
89 min_size=2,
90 max_size=10,
91 command_timeout=60.0,
92 max_retries=3,
93 initial_delay=1.0,
94 max_delay=8.0,
95 )
97 # Create storage instances
98 return GDPRStorage(
99 user_profiles=PostgresUserProfileStore(pool),
100 preferences=PostgresPreferencesStore(pool),
101 consents=PostgresConsentStore(pool),
102 conversations=PostgresConversationStore(pool),
103 audit_logs=PostgresAuditLogStore(pool),
104 )
107def create_memory_storage() -> GDPRStorage:
108 """
109 Create in-memory GDPR storage (DEVELOPMENT/TESTING ONLY)
111 WARNING: Data is lost when process restarts.
112 NOT suitable for production use.
114 Returns:
115 GDPRStorage instance with in-memory backends
117 Example:
118 >>> storage = create_memory_storage()
119 >>> # Use for testing only
120 """
121 return GDPRStorage(
122 user_profiles=InMemoryUserProfileStore(),
123 preferences=InMemoryPreferencesStore(),
124 consents=InMemoryConsentStore(),
125 conversations=InMemoryConversationStore(),
126 audit_logs=InMemoryAuditLogStore(),
127 )
130async def create_gdpr_storage(
131 backend: Literal["postgres", "memory"] = "postgres",
132 postgres_url: str = "postgresql://postgres:postgres@localhost:5432/gdpr",
133) -> GDPRStorage:
134 """
135 Create GDPR storage with specified backend
137 Factory function that creates appropriate storage backend based on configuration.
139 Args:
140 backend: Storage backend type ("postgres" or "memory")
141 postgres_url: PostgreSQL connection URL (required if backend="postgres")
143 Returns:
144 GDPRStorage instance
146 Raises:
147 ValueError: If backend is invalid
148 asyncpg.PostgresError: If PostgreSQL connection fails
150 Example:
151 >>> # Production (PostgreSQL)
152 >>> storage = await create_gdpr_storage(
153 ... backend="postgres",
154 ... postgres_url="postgresql://user:pass@db.example.com:5432/gdpr"
155 ... )
157 >>> # Development (in-memory)
158 >>> storage = await create_gdpr_storage(backend="memory")
159 """
160 if backend == "postgres": 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 return await create_postgres_storage(postgres_url)
162 elif backend == "memory": 162 ↛ 165line 162 didn't jump to line 165 because the condition on line 162 was always true
163 return create_memory_storage()
164 else:
165 raise ValueError( # noqa: TRY003
166 f"Invalid GDPR storage backend: {backend}. Must be 'postgres' or 'memory'." # noqa: EM102
167 )
170# ============================================================================
171# Global Singleton Pattern (DEPRECATED - for backward compatibility only)
172# ============================================================================
173# WARNING: Global singleton causes pytest-xdist cross-worker pollution
174# NEW CODE: Use get_gdpr_storage_dependency() for request-scoped storage
175# See: CODEX_FINDINGS_VALIDATION_REPORT_2025-11-21.md
176# ============================================================================
178_gdpr_storage: GDPRStorage | None = None
181async def initialize_gdpr_storage(
182 backend: Literal["postgres", "memory"] = "postgres",
183 postgres_url: str = "postgresql://postgres:postgres@localhost:5432/gdpr",
184) -> None:
185 """
186 Initialize global GDPR storage instance
188 .. deprecated:: 2025-11-22
189 Use get_gdpr_storage_dependency() for request-scoped storage instead.
190 This function will be removed in v2.0.
192 Should be called during application startup.
194 Args:
195 backend: Storage backend type
196 postgres_url: PostgreSQL connection URL
198 Example:
199 >>> # DEPRECATED - Old pattern
200 >>> await initialize_gdpr_storage(
201 ... backend=settings.gdpr_storage_backend,
202 ... postgres_url=settings.gdpr_postgres_url
203 ... )
204 >>>
205 >>> # NEW pattern - Request-scoped dependency injection
206 >>> # No initialization needed - storage created per-request
207 """
208 global _gdpr_storage
209 _gdpr_storage = await create_gdpr_storage(backend, postgres_url)
212def get_gdpr_storage() -> GDPRStorage:
213 """
214 Get global GDPR storage instance
216 .. deprecated:: 2025-11-22
217 Use get_gdpr_storage_dependency() for request-scoped storage instead.
218 This function will be removed in v2.0.
220 Returns:
221 GDPRStorage instance
223 Raises:
224 RuntimeError: If storage not initialized
226 Example:
227 >>> # DEPRECATED - Old pattern
228 >>> storage = get_gdpr_storage()
229 >>> profile = await storage.user_profiles.get("user:alice")
230 >>>
231 >>> # NEW pattern - FastAPI dependency injection
232 >>> from fastapi import Depends
233 >>> def my_endpoint(storage: GDPRStorage = Depends(get_gdpr_storage_dependency)):
234 ... profile = await storage.user_profiles.get("user:alice")
235 """
236 if _gdpr_storage is None:
237 msg = "GDPR storage not initialized. Call initialize_gdpr_storage() during application startup."
238 raise RuntimeError(msg)
239 return _gdpr_storage
242def reset_gdpr_storage() -> None:
243 """
244 Reset global GDPR storage instance
246 .. deprecated:: 2025-11-22
247 Not needed with request-scoped storage - each request gets fresh instance.
248 This function will be removed in v2.0.
250 Used for testing to reset storage between tests.
252 Example:
253 >>> # DEPRECATED - Old pattern
254 >>> reset_gdpr_storage()
255 >>>
256 >>> # NEW pattern - No reset needed (request-scoped)
257 >>> # Each test request gets isolated storage instance
258 """
259 global _gdpr_storage
260 _gdpr_storage = None
263# ============================================================================
264# Request-Scoped Dependency Injection (RECOMMENDED)
265# ============================================================================
266# Benefits:
267# - No global state - eliminates pytest-xdist cross-worker pollution
268# - Better testability - each request gets isolated storage
269# - Thread-safe - no shared mutable state
270# - Cloud-native - works with serverless/multi-instance deployments
271# ============================================================================
273# Configuration for request-scoped storage (set during app startup)
274_gdpr_storage_config: dict[str, Any] | None = None
277def configure_gdpr_storage(
278 backend: Literal["postgres", "memory"] = "postgres",
279 postgres_url: str = "postgresql://postgres:postgres@localhost:5432/gdpr",
280) -> None:
281 """
282 Configure GDPR storage for request-scoped dependency injection
284 Call this during application startup to set configuration.
285 Unlike initialize_gdpr_storage(), this does NOT create storage immediately.
286 Storage is created per-request using get_gdpr_storage_dependency().
288 Args:
289 backend: Storage backend type ("postgres" or "memory")
290 postgres_url: PostgreSQL connection URL
292 Example:
293 >>> # In main.py application startup
294 >>> configure_gdpr_storage(
295 ... backend="postgres",
296 ... postgres_url="postgresql://user:pass@localhost:5432/gdpr"
297 ... )
298 >>>
299 >>> # In API endpoints (automatic via dependency injection)
300 >>> @router.get("/users/me")
301 >>> async def get_user(storage: GDPRStorage = Depends(get_gdpr_storage_dependency)):
302 ... return await storage.user_profiles.get("user:alice")
303 """
304 global _gdpr_storage_config
305 _gdpr_storage_config = {
306 "backend": backend,
307 "postgres_url": postgres_url,
308 }
311async def get_gdpr_storage_dependency() -> GDPRStorage:
312 """
313 FastAPI dependency for request-scoped GDPR storage
315 Creates a new storage instance for each request using configured settings.
316 No global state - eliminates pytest-xdist cross-worker pollution.
318 Returns:
319 GDPRStorage instance (fresh per request)
321 Raises:
322 RuntimeError: If storage not configured (call configure_gdpr_storage() first)
324 Example:
325 >>> from fastapi import Depends, APIRouter
326 >>> router = APIRouter()
327 >>>
328 >>> @router.get("/users/{user_id}")
329 >>> async def get_user(
330 ... user_id: str,
331 ... storage: GDPRStorage = Depends(get_gdpr_storage_dependency)
332 >>> ):
333 ... return await storage.user_profiles.get(user_id)
334 """
335 if _gdpr_storage_config is None: 335 ↛ 345line 335 didn't jump to line 345 because the condition on line 335 was always true
336 # Fallback to global singleton for backward compatibility
337 # This allows tests to work without migration
338 if _gdpr_storage is not None: 338 ↛ 339line 338 didn't jump to line 339 because the condition on line 338 was never true
339 return _gdpr_storage
341 # If neither is configured, use default memory storage for development
342 return create_memory_storage()
344 # Create fresh storage instance per request
345 backend = _gdpr_storage_config.get("backend", "memory")
346 postgres_url = _gdpr_storage_config.get("postgres_url", "postgresql://postgres:postgres@localhost:5432/gdpr")
348 return await create_gdpr_storage(backend=backend, postgres_url=postgres_url)