Coverage for heritrace / utils / primary_source_utils.py: 97%

1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from typing import TYPE_CHECKING, cast 

6 

7from flask import current_app 

8from redis import RedisError 

9 

10from heritrace.utils.uri_utils import is_valid_url 

11 

12if TYPE_CHECKING: 

13 from redis import Redis 

14 

15USER_DEFAULT_SOURCE_KEY = "user:{user_id}:default_primary_source" 

16 

17 

18def get_user_default_primary_source(user_id: str) -> str | None: 

19 """ 

20 Get the user's default primary source from Redis 

21 

22 Args: 

23 user_id: The user's ID (e.g. ORCID) 

24 

25 Returns: 

26 The user's default primary source or None if not set or error 

27 """ 

28 user_default_source = None 

29 if user_id: 

30 key = USER_DEFAULT_SOURCE_KEY.format(user_id=user_id) 

31 try: 

32 redis_client: Redis = current_app.extensions["redis_client"] # type: ignore[type-arg] 

33 user_default_source = cast("str | None", redis_client.get(key)) 

34 except RedisError: 

35 current_app.logger.exception( 

36 "Failed to get user default primary source from Redis" 

37 ) 

38 user_default_source = None 

39 

40 return user_default_source 

41 

42 

43def get_default_primary_source(user_id: str) -> str: 

44 """ 

45 Get the default primary source to use, falling back to app config if user has none 

46 

47 Args: 

48 user_id: The user's ID (e.g. ORCID) 

49 

50 Returns: 

51 The primary source to use (user's default or app default) 

52 """ 

53 user_default = get_user_default_primary_source(user_id) 

54 return user_default or current_app.config["PRIMARY_SOURCE"] 

55 

56 

57def save_user_default_primary_source(user_id: str, primary_source: str) -> bool | None: 

58 """ 

59 Save the user's default primary source to Redis 

60 

61 Args: 

62 user_id: The user's ID (e.g. ORCID) 

63 primary_source: The URL to save as default primary source 

64 

65 Returns: 

66 True if saved successfully, False otherwise 

67 """ 

68 if not user_id or not primary_source or not is_valid_url(primary_source): 

69 return False 

70 

71 key = USER_DEFAULT_SOURCE_KEY.format(user_id=user_id) 

72 try: 

73 redis_client: Redis = current_app.extensions["redis_client"] # type: ignore[type-arg] 

74 redis_client.set(key, primary_source) 

75 except RedisError: 

76 current_app.logger.exception( 

77 "Failed to save user default primary source to Redis" 

78 ) 

79 return False 

80 else: 

81 return True