Coverage for src / local_deep_research / database / auth_db.py: 94%

1""" 

2Authentication database initialization and management. 

3This manages the central ldr_auth.db which only stores usernames. 

4""" 

5 

6from pathlib import Path 

7 

8from loguru import logger 

9from sqlalchemy import create_engine 

10from sqlalchemy.orm import sessionmaker 

11 

12from ..config.paths import get_data_directory 

13from .models.auth import User 

14from .models.base import Base 

15 

16 

17def get_auth_db_path() -> Path: 

18 """Get the path to the authentication database.""" 

19 return get_data_directory() / "ldr_auth.db" 

20 

21 

22def init_auth_database(): 

23 """Initialize the authentication database if it doesn't exist.""" 

24 auth_db_path = get_auth_db_path() 

25 

26 # Ensure the data directory exists 

27 auth_db_path.parent.mkdir(parents=True, exist_ok=True) 

28 

29 # Check if database already exists 

30 if auth_db_path.exists(): 

31 logger.debug(f"Auth database already exists at {auth_db_path}") 

32 return 

33 

34 logger.info(f"Creating auth database at {auth_db_path}") 

35 

36 # Create the database 

37 engine = create_engine(f"sqlite:///{auth_db_path}") 

38 

39 # Create tables 

40 Base.metadata.create_all(engine, tables=[User.__table__]) 

41 

42 logger.info("Auth database initialized successfully") 

43 

44 

45def get_auth_db_session(): 

46 """Get a session for the auth database.""" 

47 auth_db_path = get_auth_db_path() 

48 

49 # Ensure database exists 

50 if not auth_db_path.exists(): 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 init_auth_database() 

52 

53 engine = create_engine(f"sqlite:///{auth_db_path}") 

54 Session = sessionmaker(bind=engine) 

55 return Session() 

56 

57 

58# Initialize on import 

59init_auth_database()