Coverage for app/backend_pre_start.py: 70%

1import logging 

2 

3from sqlalchemy import Engine 

4from sqlmodel import Session, select 

5from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed 

6 

7from app.core.db import engine 

8 

9logging.basicConfig(level=logging.INFO) 

10logger = logging.getLogger(__name__) 

11 

12max_tries = 60 * 5 # 5 minutes 

13wait_seconds = 1 

14 

15 

16@retry( 

17 stop=stop_after_attempt(max_tries), 

18 wait=wait_fixed(wait_seconds), 

19 before=before_log(logger, logging.INFO), 

20 after=after_log(logger, logging.WARN), 

21) 

22def init(db_engine: Engine) -> None: 

23 try: 

24 with Session(db_engine) as session: 

25 # Try to create session to check if DB is awake 

26 session.exec(select(1)) 

27 except Exception as e: 

28 logger.error(e) 

29 raise e 

30 

31 

32def main() -> None: 

33 logger.info("Initializing service") 

34 init(engine) 

35 logger.info("Service finished initializing") 

36 

37 

38if __name__ == "__main__": 

39 main()