Coverage for app/services/advisory_lock.py: 100%

1""" 

2PostgreSQL advisory-lock helper for background jobs that must run once per 

3scheduled tick rather than once per gunicorn worker. 

4 

5Production runs `gunicorn --workers 4` without `--preload` 

6(docker/docker-entrypoint.sh), so every scheduler thread started in 

7create_app() runs independently in all four worker processes. Jobs guarded 

8here acquire a well-known lock id before doing their work; the first worker 

9to grab it proceeds, the rest skip that run and try again on the next tick. 

10 

11The lock is held on a dedicated connection (not the ORM session's pooled 

12connection), so it survives however many commits the guarded work performs — 

13acquiring it via the session and relying on `pg_try_advisory_xact_lock` 

14would release the lock at the *first* commit, letting a second worker start 

15racing partway through a multi-commit job. 

16 

17Lock id registry (pick a new one when adding a caller): 

18 7283910456 — welcome email, startup, one-shot (services/notification_service.py) 

19 7283910457 — daily notification checks (services/notification_service.py) 

20 7283910458 — EASA airworthiness sync (airworthiness_sync.py) 

21 7283910459 — document sync-watcher scan (sync_watcher.py) 

22 7283910460 — scheduled backup + retention (services/backup_scheduler.py) 

23""" 

24 

25from collections.abc import Iterator 

26from contextlib import contextmanager 

27from typing import Any 

28 

29 

30@contextmanager 

31def advisory_lock_scope(db: Any, lock_id: int) -> Iterator[bool]: 

32 """Yield True if lock_id was acquired for the duration of the `with` block. 

33 

34 On non-PostgreSQL engines (SQLite in dev/test), always yields True 

35 without touching the database. 

36 """ 

37 if db.engine.dialect.name != "postgresql": 

38 yield True 

39 return 

40 

41 from sqlalchemy import text # pyright: ignore[reportMissingImports] 

42 

43 conn = db.engine.connect() 

44 try: 

45 acquired = bool( 

46 conn.execute( 

47 text("SELECT pg_try_advisory_lock(:lock_id)"), {"lock_id": lock_id} 

48 ).scalar() 

49 ) 

50 try: 

51 yield acquired 

52 finally: 

53 if acquired: 

54 conn.execute( 

55 text("SELECT pg_advisory_unlock(:lock_id)"), {"lock_id": lock_id} 

56 ) 

57 finally: 

58 conn.close()