Coverage for application / qaqc / tator / util.py: 19%

1import json 

2 

3import tator 

4import requests 

5from flask import current_app, flash, redirect, session 

6 

7 

8def init_tator_api(): 

9 if 'tator_token' not in session: 

10 flash('Please log in to Tator', 'info') 

11 return None, redirect('/') 

12 try: 

13 api = tator.get_api( 

14 host=current_app.config.get('TATOR_URL'), 

15 token=session['tator_token'], 

16 ) 

17 return api, None 

18 except tator.openapi.tator_openapi.exceptions.ApiException as e: 

19 flash(json.loads(e.body)['message'], 'danger') 

20 return None, redirect('/') 

21 

22 

23# TODO - either update review server to allow batch requests by sequence, or use threading to make this faster 

24def get_comments_and_image_refs(deployment_names: list[str]) -> tuple[dict, dict]: 

25 comments = {} 

26 image_refs = {} 

27 try: 

28 for deployment in deployment_names: 

29 print(f'Getting comments for deployment {deployment}...') 

30 comment_res = requests.get( 

31 url=f'{current_app.config.get("DARC_REVIEW_URL")}/comment/sequence/{deployment}', 

32 headers=current_app.config.get('DARC_REVIEW_HEADERS'), 

33 ) 

34 if comment_res.status_code != 200: 

35 raise requests.exceptions.ConnectionError 

36 comments |= comment_res.json() 

37 image_ref_res = requests.get(f'{current_app.config.get("DARC_REVIEW_URL")}/image-reference/quick') 

38 if image_ref_res.status_code != 200: 

39 raise requests.exceptions.ConnectionError 

40 image_refs = image_ref_res.json() 

41 except requests.exceptions.ConnectionError: 

42 print('\nERROR: unable to connect to external review server\n') 

43 return comments, image_refs