Coverage for src/lexigram/web/integrations/sql.py: 50%

1"""SQL integration — attaches DB pool to ASGI app state for lifespan cleanup.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.contracts.core.di import ContainerResolverProtocol 

8from lexigram.contracts.data import DatabaseProviderProtocol 

9from lexigram.contracts.exceptions import UnresolvableDependencyError 

10from lexigram.contracts.exceptions.provider import ModuleVisibilityError 

11from lexigram.logging import get_logger 

12 

13if TYPE_CHECKING: 

14 from starlette.applications import Starlette 

15 

16logger = get_logger(__name__) 

17 

18 

19class SQLIntegration: 

20 """Resolves the primary DB pool from the container and attaches it to app.state. 

21 

22 Optional integration: silently skips if no SQL module is registered. This 

23 preserves the framework contract that web can run without a database. 

24 """ 

25 

26 @staticmethod 

27 async def configure( 

28 app: Starlette, 

29 container: ContainerResolverProtocol, 

30 ) -> None: 

31 """Attach the primary database pool to app.state.db_pool. 

32 

33 Args: 

34 app: The ASGI application. 

35 container: The DI container resolver. 

36 

37 Raises: 

38 LookupError: If no SQL module is registered (silent skip). 

39 """ 

40 try: 

41 db_provider = await container.resolve(DatabaseProviderProtocol) 

42 except (LookupError, UnresolvableDependencyError, ModuleVisibilityError): 

43 logger.debug("No SQL module registered; SQLIntegration skipped") 

44 return 

45 

46 if db_provider is None: 

47 logger.debug("No SQL module registered; SQLIntegration skipped") 

48 return 

49 

50 pool = await db_provider.get_primary_pool() 

51 app.state.db_pool = pool 

52 logger.info("SQLIntegration: db_pool attached to app.state")