Coverage for src/backend/InvenTree/InvenTree/exchange.py: 65%

1"""Custom exchange backend which hooks into the InvenTree plugin system to fetch exchange rates from an external API.""" 

2 

3from django.db.transaction import atomic 

4 

5import structlog 

6from djmoney.contrib.exchange.backends.base import SimpleExchangeBackend 

7from djmoney.contrib.exchange.models import ExchangeBackend, Rate 

8 

9from common.currency import currency_code_default, currency_codes 

10from common.settings import get_global_setting 

11 

12logger = structlog.get_logger('inventree') 

13 

14 

15class InvenTreeExchange(SimpleExchangeBackend): 

16 """Backend for automatically updating currency exchange rates. 

17 

18 Uses the plugin system to actually fetch the rates from an external API. 

19 """ 

20 

21 name = 'InvenTreeExchange' 

22 

23 def get_rates(self, **kwargs) -> dict: 

24 """Set the requested currency codes and get rates.""" 

25 from plugin import PluginMixinEnum, registry 

26 

27 base_currency = kwargs.get('base_currency', currency_code_default()) 

28 symbols = kwargs.get('symbols', currency_codes()) 

29 

30 # Find the selected exchange rate plugin 

31 slug = get_global_setting('CURRENCY_UPDATE_PLUGIN', create=False) 

32 

33 plugin = registry.get_plugin(slug, active=True) if slug else None 

34 

35 if not plugin: 35 ↛ 37line 35 didn't jump to line 37 because the condition on line 35 was never true

36 # Find the first active currency exchange plugin 

37 plugins = registry.with_mixin( 

38 PluginMixinEnum.CURRENCY_EXCHANGE, active=True 

39 ) 

40 

41 if len(plugins) > 0: 

42 plugin = plugins[0] 

43 

44 if not plugin: 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true

45 logger.warning( 

46 'No active currency exchange plugins found - skipping update' 

47 ) 

48 return {} 

49 

50 logger.info("Running exchange rate update using plugin '%s'", plugin.name) 

51 

52 # Plugin found - run the update task 

53 try: 

54 rates = plugin.update_exchange_rates(base_currency, symbols) 

55 except Exception as exc: 

56 logger.exception('Exchange rate update failed: %s', exc) 

57 return {} 

58 

59 if not rates: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 logger.warning( 

61 'Exchange rate update failed - no data returned from plugin %s', slug 

62 ) 

63 return {} 

64 

65 # Update exchange rates based on returned data 

66 if type(rates) is not dict: 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 logger.warning( 

68 'Invalid exchange rate data returned from plugin %s (type %s)', 

69 slug, 

70 type(rates), 

71 ) 

72 return {} 

73 

74 # Ensure base currency is provided 

75 rates[base_currency] = 1.00 

76 

77 return rates 

78 

79 @atomic 

80 def update_rates(self, base_currency=None, **kwargs): 

81 """Call to update all exchange rates.""" 

82 backend, _ = ExchangeBackend.objects.update_or_create( 

83 name=self.name, defaults={'base_currency': base_currency} 

84 ) 

85 

86 if base_currency is None: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 base_currency = currency_code_default() 

88 

89 symbols = currency_codes() 

90 

91 logger.info( 

92 'Updating exchange rates for %s (%s currencies)', 

93 base_currency, 

94 len(symbols), 

95 ) 

96 

97 # Fetch new rates from the backend 

98 # If the backend fails, the existing rates will not be updated 

99 rates = self.get_rates(base_currency=base_currency, symbols=symbols) 

100 

101 if rates: 101 ↛ 110line 101 didn't jump to line 110 because the condition on line 101 was always true

102 # Clear out existing rates 

103 backend.clear_rates() 

104 

105 Rate.objects.bulk_create([ 

106 Rate(currency=currency, value=amount, backend=backend) 

107 for currency, amount in rates.items() 

108 ]) 

109 else: 

110 logger.info( 

111 'No exchange rates returned from backend - currencies not updated' 

112 ) 

113 

114 logger.info('Updated exchange rates for %s', base_currency)