Coverage for src / idx_api / utils / cache.py: 22%

1"""Cache invalidation utilities for coordinating with Astro frontend. 

2 

3When brokerage settings change, the Astro SSR server needs to be notified 

4to clear its in-memory config cache. This module provides async functions 

5to call the Astro purge endpoint. 

6""" 

7 

8import os 

9import httpx 

10from typing import Optional 

11 

12# Astro frontend URL for internal service-to-service communication 

13ASTRO_INTERNAL_URL = os.getenv("ASTRO_INTERNAL_URL", "http://idx-web:4321") 

14INTERNAL_API_KEY = os.getenv("INTERNAL_API_KEY", "idx-internal-key") 

15 

16 

17async def invalidate_site_config(domain: Optional[str] = None) -> bool: 

18 """ 

19 Notify Astro frontend to purge cached site config. 

20 

21 Args: 

22 domain: Optional specific domain to purge. If None, purges all cached configs. 

23 

24 Returns: 

25 True if purge was successful, False otherwise. 

26 """ 

27 try: 

28 async with httpx.AsyncClient(timeout=5.0) as client: 

29 response = await client.post( 

30 f"{ASTRO_INTERNAL_URL}/api/internal/purge-config", 

31 headers={ 

32 "Authorization": f"Bearer {INTERNAL_API_KEY}", 

33 "Content-Type": "application/json", 

34 }, 

35 json={"domain": domain} if domain else {}, 

36 ) 

37 

38 if response.status_code == 200: 

39 result = response.json() 

40 print(f"✅ Cache invalidated: {result}") 

41 return True 

42 else: 

43 print(f"⚠️ Cache invalidation failed: {response.status_code} - {response.text}") 

44 return False 

45 

46 except httpx.ConnectError: 

47 # Astro server might not be running (dev mode) - not critical 

48 print("⚠️ Could not connect to Astro server for cache invalidation (may be in dev mode)") 

49 return False 

50 except Exception as e: 

51 print(f"⚠️ Cache invalidation error: {e}") 

52 return False 

53 

54 

55async def invalidate_brokerage_domains(brokerage_id: int, db) -> bool: 

56 """ 

57 Invalidate cache for all domains associated with a brokerage. 

58 

59 Args: 

60 brokerage_id: The brokerage ID whose domains should be purged. 

61 db: Database session to query domains. 

62 

63 Returns: 

64 True if all purges were successful. 

65 """ 

66 from idx_api.models.brokerage_domain import BrokerageDomain 

67 from sqlalchemy import select 

68 

69 # Get all domains for this brokerage 

70 domains = db.scalars( 

71 select(BrokerageDomain.domain) 

72 .where(BrokerageDomain.brokerage_id == brokerage_id) 

73 ).all() 

74 

75 if not domains: 

76 # No domains configured - purge all just in case 

77 return await invalidate_site_config() 

78 

79 # Purge each domain's cache 

80 success = True 

81 for domain in domains: 

82 result = await invalidate_site_config(domain) 

83 success = success and result 

84 

85 return success