Coverage for django_napse/core/db_essentials.py: 90%

1import json 

2 

3from django.apps import apps 

4from django.db.models.signals import post_migrate 

5from django.db.transaction import atomic 

6from django.dispatch import receiver 

7 

8from django_napse.core.settings import napse_settings as settings 

9 

10 

11@receiver(post_migrate) 

12def create_exchanges(sender, **kwargs): 

13 Exchange = apps.get_model("django_napse_core", "Exchange") 

14 exchange_configs = settings.NAPSE_EXCHANGE_CONFIGS 

15 created_exchanges = [] 

16 with atomic(): 

17 for exchange_name, exchange_config in exchange_configs.items(): 

18 _, created = Exchange.objects.get_or_create(name=exchange_name, description=exchange_config["description"]) 

19 if created: 

20 created_exchanges.append(exchange_name) 

21 if len(created_exchanges) > 0: 

22 print(f"Created exchanges: {', '.join(created_exchanges)}") 

23 

24 

25@receiver(post_migrate) 

26def create_accounts(sender, **kwargs): 

27 Exchange = apps.get_model("django_napse_core", "Exchange") 

28 from django_napse.core.models.accounts.exchange import EXCHANGE_ACCOUNT_DICT 

29 

30 try: 

31 with open(settings.NAPSE_SECRETS_FILE_PATH, "r") as json_file: 

32 secrets = json.load(json_file) 

33 except FileNotFoundError: 

34 with open(settings.NAPSE_SECRETS_FILE_PATH, "w") as json_file: 

35 json.dump({"Exchange Accounts": {}}, json_file) 

36 secrets = {"Exchange Accounts": {}} 

37 created_exchange_accounts = [] 

38 

39 with atomic(): 

40 for exchange_id, exchange_secrets in secrets["Exchange Accounts"].items(): 

41 exchange_name = exchange_secrets.pop("exchange") 

42 exchange = Exchange.objects.get(name=exchange_name) 

43 

44 try: 

45 EXCHANGE_ACCOUNT_DICT[exchange_name].objects.get( 

46 name=exchange_id, 

47 exchange=exchange, 

48 **exchange_secrets, 

49 ) 

50 except EXCHANGE_ACCOUNT_DICT[exchange_name].DoesNotExist: 

51 EXCHANGE_ACCOUNT_DICT[exchange_name].objects.create( 

52 name=exchange_id, 

53 exchange=exchange, 

54 default=True, 

55 **exchange_secrets, 

56 ) 

57 created_exchange_accounts.append(exchange_id) 

58 

59 if len(created_exchange_accounts) > 0: 

60 print(f"Created exchange accounts: {', '.join(created_exchange_accounts)}")