Coverage for heritrace/utils/primary_source_utils.py: 100%
28 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-06-24 11:39 +0000
« prev ^ index » next coverage.py v7.6.12, created at 2025-06-24 11:39 +0000
1from flask import current_app
2import validators
4USER_DEFAULT_SOURCE_KEY = "user:{user_id}:default_primary_source"
6def get_user_default_primary_source(user_id):
7 """
8 Get the user's default primary source from Redis
10 Args:
11 user_id: The user's ID (e.g. ORCID)
13 Returns:
14 The user's default primary source or None if not set or error
15 """
16 user_default_source = None
17 if user_id:
18 key = USER_DEFAULT_SOURCE_KEY.format(user_id=user_id)
19 try:
20 user_default_source = current_app.redis_client.get(key)
21 if user_default_source and isinstance(user_default_source, bytes):
22 user_default_source = user_default_source.decode('utf-8')
23 except Exception as e:
24 current_app.logger.error(f"Failed to get user default primary source from Redis: {e}")
25 user_default_source = None # Ensure it's None on error
27 return user_default_source
29def get_default_primary_source(user_id):
30 """
31 Get the default primary source to use, falling back to app config if user has none
33 Args:
34 user_id: The user's ID (e.g. ORCID)
36 Returns:
37 The primary source to use (user's default or app default)
38 """
39 user_default = get_user_default_primary_source(user_id)
40 return user_default or current_app.config["PRIMARY_SOURCE"]
42def save_user_default_primary_source(user_id, primary_source):
43 """
44 Save the user's default primary source to Redis
46 Args:
47 user_id: The user's ID (e.g. ORCID)
48 primary_source: The URL to save as default primary source
50 Returns:
51 True if saved successfully, False otherwise
52 """
53 if not user_id or not primary_source or not validators.url(primary_source):
54 return False
56 key = USER_DEFAULT_SOURCE_KEY.format(user_id=user_id)
57 try:
58 current_app.redis_client.set(key, primary_source)
59 return True
60 except Exception as e:
61 current_app.logger.error(f"Failed to save user default primary source to Redis: {e}")
62 return False