Belle II Software light-2406-ragdoll
rundb.py
1#!/usr/bin/env python3
2
3
10
11'''
12rundb - Helper classes for retrieving information from the RunDB
13----------------------------------------------------------------
14
15This modules contains classes useful to deal with the RunDB:
16
17* `RunDB`, a simple API class to just get run information from the RunDB
18'''
19
20import requests
21import getpass
22
23
24class RunDB:
25 """
26 Simple API class to just get run information from the RunDB.
27
28 Parameters:
29 apikey (str): RunDB API key (see
30 `this question <https://questions.belle2.org/question/11145/obtaining-an-access-token-for-rundb-api-2021-version/>`_
31 or `this Confluence page <https://confluence.desy.de/x/DZ8CCg>`_ how to get one)
32 username (str): DESY username
33 """
34
35
36 URL = "https://rundb.belle2.org"
37
38 def __init__(self, apikey=None, username=None):
39 """Create an object and setup authentication."""
40
41 self._session = requests.Session()
42 if apikey is None:
43 # If no specific username use the local system username
44 if username is None:
45 username = getpass.getuser()
46 # If we don't have an api key use desy username/password
47 self._session.auth = (username, getpass.getpass(f"DESY password ({username}): "))
48 else:
49 # Otherwise use the api key
50 self._session.headers.update({'Authorization': f'Bearer {apikey}'})
51 # And request json output ...
52 self._session.headers.update({'Content-Type': 'application/json'})
53
54 def _pagination(self, request):
55 """Deal with API pagination of an initial request to the API.
56
57 It will return all the objects from all pages lazily requesting new pages
58 as objects are consumed. Will work for all list requests to the server.
59 """
60 while True:
61 # check the return value and raise exception on error
62 request.raise_for_status()
63 # and otherwise get the json
64 result = request.json()
65 # and return the objects one by one by yielding objects from the list
66 # of results
67 yield from result['results']
68 # check if there's a next page, if not done
69 if result['next'] is None:
70 break
71 # otherwise continue with the next page
72 # yees, global variable ...
73 request = self._session.get(result['next'])
74
75 def get_run_info(self, **search_params):
76 """Return the run information from the run registry.
77
78 All arguments are forwarded to the run registry ``/run/`` method
79 documented at the following `link <https://rundb.belle2.org/rest/v1/swagger/>`_.
80 Please check there for up to date documentation, at the time of
81 this writing the supported arguments are:
82
83 * min_experiment (int)
84 * min_run (int)
85 * max_experiment (int)
86 * max_run (int)
87 * min_date (iso8601 date string, e.g. 2020-05-06)
88 * max_date (iso8601 date string, e.g. 2020-05-06)
89 * all_detectors_running (bool)
90 * expand (bool): If true return full run objects, not just a summary
91 links to the run objects
92
93 If ``expand=False`` you can request the full objects for each run by calling
94 `get_details` with the returned run summary object as argument.
95 ``expand=False`` is much faster if no further details are needed but
96 getting the details in a separate step for many many runs will be slow
97 so depending on how many runs are selected one or the other may be
98 faster.
99 """
100 req = self._session.get(f'{self.URL}/rest/v1/runs/', params=search_params)
101 return self._pagination(req)
102
103 def get_details(self, run_summary):
104 """
105 Return details for a run summary object returned from `get_run_info`
106 if ``expand`` was not set to True
107
108 Parameters:
109 run_summary: a run summary object returned from `get_run_info`
110 """
111 # Get the url object
112 req = self._session.get(run_summary['url'])
113 # Raise an exception in case of any error
114 req.raise_for_status()
115 # And return the json object
116 return req.json()
def get_details(self, run_summary)
Definition: rundb.py:103
def get_run_info(self, **search_params)
Definition: rundb.py:75
def _pagination(self, request)
Definition: rundb.py:54
_session
session object for connection to the RunDB
Definition: rundb.py:41
def __init__(self, apikey=None, username=None)
Definition: rundb.py:38