Coverage for dormatory/models/database_config.py: 52%

1""" 

2Database configuration for DORMATORY. 

3 

4This module provides database configuration utilities for different database types 

5(SQLite and PostgreSQL) and handles database-specific requirements. 

6""" 

7 

8import os 

9from typing import Optional 

10from sqlalchemy import create_engine 

11from sqlalchemy.orm import sessionmaker 

12 

13 

14def get_database_url() -> str: 

15 """ 

16 Get database URL from environment variable or use default. 

17  

18 Returns: 

19 Database URL string 

20 """ 

21 return os.getenv("DATABASE_URL", "sqlite:///dormatory.db") 

22 

23 

24def create_engine_and_session(database_url: Optional[str] = None): 

25 """ 

26 Create SQLAlchemy engine and session factory. 

27  

28 Args: 

29 database_url: Database connection URL. If None, uses environment or default. 

30  

31 Returns: 

32 Tuple of (engine, SessionLocal) 

33 """ 

34 if database_url is None: 

35 database_url = get_database_url() 

36 

37 # Configure engine based on database type 

38 if database_url.startswith("sqlite"): 

39 # SQLite configuration 

40 engine = create_engine( 

41 database_url, 

42 echo=True, 

43 connect_args={"check_same_thread": False} 

44 ) 

45 elif database_url.startswith("postgresql"): 

46 # PostgreSQL configuration 

47 engine = create_engine( 

48 database_url, 

49 echo=True, 

50 pool_pre_ping=True, 

51 pool_recycle=300 

52 ) 

53 else: 

54 # Default configuration 

55 engine = create_engine(database_url, echo=True) 

56 

57 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) 

58 return engine, SessionLocal 

59 

60 

61def get_database_info() -> dict: 

62 """ 

63 Get information about the current database configuration. 

64  

65 Returns: 

66 Dictionary with database information 

67 """ 

68 database_url = get_database_url() 

69 

70 if database_url.startswith("sqlite"): 

71 db_type = "SQLite" 

72 db_name = database_url.replace("sqlite:///", "").replace("sqlite://", "") 

73 elif database_url.startswith("postgresql"): 

74 db_type = "PostgreSQL" 

75 # Extract database name from PostgreSQL URL 

76 db_name = database_url.split("/")[-1].split("?")[0] 

77 else: 

78 db_type = "Unknown" 

79 db_name = database_url 

80 

81 return { 

82 "type": db_type, 

83 "name": db_name, 

84 "url": database_url 

85 } 

86 

87 

88def is_postgresql() -> bool: 

89 """ 

90 Check if the current database is PostgreSQL. 

91  

92 Returns: 

93 True if PostgreSQL, False otherwise 

94 """ 

95 return get_database_url().startswith("postgresql") 

96 

97 

98def is_sqlite() -> bool: 

99 """ 

100 Check if the current database is SQLite. 

101  

102 Returns: 

103 True if SQLite, False otherwise 

104 """ 

105 return get_database_url().startswith("sqlite")