Coverage for atlas_ecommerce/config.py: 94%

1"""Settings and secrets. 

2 

3Mirrors auth-proxy: plain config in ``Settings``, anything sensitive in 

4``Secrets``, resolved by atlas-tools from env, then ``.env.{environment}``, then 

5GCP Secret Manager. The project root is passed explicitly because atlas-tools 

6would otherwise infer it from its own installed location. 

7""" 

8 

9from functools import lru_cache 

10from pathlib import Path 

11 

12from async_lru import alru_cache 

13from atlas_tools.settings import ManagedSecrets, load_secrets, select_env_file 

14from pydantic import AliasChoices, AliasGenerator, Field, SecretStr 

15from pydantic_settings import BaseSettings, SettingsConfigDict 

16 

17PROJECT_ROOT = Path(__file__).resolve().parent.parent 

18 

19 

20def ae_alias(field_name: str) -> AliasChoices: 

21 """Accept a setting bare or AE_-prefixed. 

22 

23 Secrets are always AE_-prefixed, so allowing settings to carry it too lets a 

24 whole .env use one prefix. Lookup is case-insensitive, so this matches 

25 AE_MONGO_DB_NAME. 

26 """ 

27 return AliasChoices(field_name, f"ae_{field_name}") 

28 

29 

30class Settings(BaseSettings): 

31 model_config = SettingsConfigDict( 

32 extra="ignore", 

33 frozen=True, 

34 alias_generator=AliasGenerator(validation_alias=ae_alias), 

35 populate_by_name=True, 

36 ) 

37 

38 project_id: str = Field(default="sample_project_id") 

39 environment: str = Field(default="dev") 

40 

41 mongo_db_name: str = Field(default="atlas_ecommerce") 

42 auth_proxy_base_url: str = Field(default="http://localhost:8004") 

43 

44 # Shopify topic this service is subscribed to. Payment is taken upfront, so 

45 # orders/paid lands within seconds of orders/create while still guaranteeing 

46 # captured money. Switching to orders/create is a config change only. 

47 shopify_topic: str = Field(default="orders/paid") 

48 # Shop domains deliveries may originate from, comma-separated; empty 

49 # disables the check. A store can have more than one myshopify.com domain - 

50 # a customer-facing primary and the canonical one Shopify identifies it by - 

51 # and X-Shopify-Shop-Domain carries the canonical one, which is not always 

52 # the domain the storefront is configured with. 

53 shopify_shop_domain: str = Field(default="") 

54 

55 @property 

56 def shopify_shop_domains(self) -> frozenset[str]: 

57 """The accepted shop domains, lower-cased. Empty means accept any.""" 

58 return frozenset(d.strip().lower() for d in self.shopify_shop_domain.split(",") if d.strip()) 

59 

60 provision_max_attempts: int = Field(default=8) 

61 provision_batch_size: int = Field(default=50) 

62 provision_poll_seconds: float = Field(default=5.0) 

63 

64 # Purchase conversions fanned out from orders/paid. A destination with an empty 

65 # setting (or missing secrets) is not delivered from this process; its dispatches 

66 # wait, pending, for a replica that has the client. 

67 linkedin_conversion_urn: str = Field(default="") 

68 linkedin_api_version: str = Field(default="202608") 

69 x_pixel_id: str = Field(default="") 

70 x_purchase_event_id: str = Field(default="") 

71 mixpanel_token: str = Field(default="") 

72 meta_pixel_id: str = Field(default="") 

73 meta_api_version: str = Field(default="v26.0") 

74 meta_test_event_code: str = Field(default="") 

75 google_ads_customer_id: str = Field(default="") 

76 google_ads_login_customer_id: str = Field(default="") 

77 google_ads_purchase_conversion_action: str = Field(default="") 

78 google_data_manager_validate_only: bool = Field(default=False) 

79 # With the worker's backoff (30 s doubling, capped at an hour) thirty attempts span about a day. 

80 conversion_max_attempts: int = Field(default=30) 

81 conversion_batch_size: int = Field(default=50) 

82 conversion_poll_seconds: float = Field(default=5.0) 

83 conversion_lease_seconds: float = Field(default=120.0) 

84 

85 

86class Secrets(ManagedSecrets): 

87 model_config = SettingsConfigDict(env_prefix="AE_", extra="ignore", frozen=True) 

88 

89 gcp_secret_name_templates = ("{prefix}{field}",) 

90 bare_env_fallback = True 

91 

92 # Shopify app client secret. Webhooks registered through the admin UI are 

93 # signed with a different key than app-registered ones; this must match 

94 # whichever route registered the subscription. 

95 shopify_webhook_secret: SecretStr = Field(default=...) 

96 mongo_uri: SecretStr = Field(default=...) 

97 logfire_token: SecretStr | None = Field(default=None) 

98 

99 linkedin_client_id: SecretStr | None = Field(default=None) 

100 linkedin_client_secret: SecretStr | None = Field(default=None) 

101 linkedin_refresh_token: SecretStr | None = Field(default=None) 

102 linkedin_access_token: SecretStr | None = Field(default=None) 

103 x_pixel_token: SecretStr | None = Field(default=None) 

104 meta_capi_token: SecretStr | None = Field(default=None) 

105 google_ads_client_id: SecretStr | None = Field(default=None) 

106 google_ads_client_secret: SecretStr | None = Field(default=None) 

107 google_ads_refresh_token: SecretStr | None = Field(default=None) 

108 

109 

110class LogfireSettings(BaseSettings): 

111 """The Logfire token on its own, resolvable without an event loop. 

112 

113 Instrumentation has to be applied before the server starts serving, which is 

114 before the lifespan can await ``Secrets``. Every deployed environment 

115 injects the token as an env var (``envFrom: secretRef``) and local runs read 

116 ``.env.{environment}``, so a sync read covers both; GCP Secret Manager is 

117 deliberately not consulted here. 

118 """ 

119 

120 model_config = SettingsConfigDict(env_prefix="AE_", extra="ignore", frozen=True) 

121 

122 logfire_token: SecretStr | None = Field(default=None) 

123 

124 

125def env_file() -> Path | None: 

126 """The .env file for the current ENVIRONMENT, or None.""" 

127 return select_env_file(PROJECT_ROOT) 

128 

129 

130@lru_cache 

131def get_settings() -> Settings: 

132 selected = env_file() 

133 return Settings(_env_file=selected) if selected else Settings() 

134 

135 

136@lru_cache 

137def get_logfire_token() -> str | None: 

138 """The Logfire token, or None when unset.""" 

139 selected = env_file() 

140 settings = LogfireSettings(_env_file=selected) if selected else LogfireSettings() 

141 return settings.logfire_token.get_secret_value() if settings.logfire_token else None 

142 

143 

144@alru_cache 

145async def get_secrets() -> Secrets: 

146 """Load secrets via atlas-tools: init > env > .env > GCP Secret Manager.""" 

147 return await load_secrets(Secrets, _env_file=env_file())