ConditionsDB Class Reference

Classes
class	RequestError

Public Member Functions
def	__init__ (self, base_url=None, max_connections=10, retries=3)
def	set_authentication_token (self, token)
def	request (self, method, url, message=None, *args, **argk)
def	get_globalTags (self)
def	has_globalTag (self, name)
def	get_globalTagInfo (self, name)
def	get_globalTagType (self, name)
def	create_globalTag (self, name, description, user)
def	get_all_iovs (self, globalTag, exp=None, run=None, message=None, run_range=None, fully_contained=False)
def	get_payloads (self, global_tag=None)
def	check_payloads (self, payloads, information="payloadId")
def	get_revisions (self, entries)
def	create_payload (self, module, filename, checksum=None)
def	create_iov (self, globalTagId, payloadId, firstExp, firstRun, finalExp, finalRun)
def	delete_iov (self, iovId)
def	modify_iov (self, iovId, firstExp, firstRun, finalExp, finalRun)
def	get_iovs (self, globalTagName, payloadName=None)
def	upload (self, filename, global_tag, normalize=False, ignore_existing=False, nprocess=1, uploaded_entries=None)
def	staging_request (self, filename, normalize, data, password)

Static Public Member Functions
def	get_base_urls (given_url)

Static Public Attributes
list	BASE_URLS = [conditions.default_metadata_provider_url]
	base url to the conditions db to be used if no custom url is given

Protected Attributes
	_session
	session object to get keep-alive support and connection pooling
	_base_url
	base url to be prepended to all requests

Detailed Description

Class to interface conditions db REST interface

Definition at line 231 of file __init__.py.

Constructor & Destructor Documentation

__init__()

def __init__	(		self,
			base_url = None,
			max_connections = 10,
			retries = 3
	)

Create a new instance of the interface

Args:
    base_url (string): base url of the rest interface
    max_connections (int): number of connections to keep open, mostly useful for threaded applications
    retries (int): number of retries in case of connection problems

Definition at line 273 of file __init__.py.

    def __init__(self, base_url=None, max_connections=10, retries=3):
        """
        Create a new instance of the interface
 
        Args:
            base_url (string): base url of the rest interface
            max_connections (int): number of connections to keep open, mostly useful for threaded applications
            retries (int): number of retries in case of connection problems
        """
 
        
        self._session = requests.Session()
        # and set the connection options we want to have
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=max_connections, pool_maxsize=max_connections,
            max_retries=retries, pool_block=True
        )
        self._session.mount("http://", adapter)
        self._session.mount("https://", adapter)
        # and also set the proxy settings
        if "BELLE2_CONDB_PROXY" in os.environ:
            self._session.proxies = {
                "http": os.environ.get("BELLE2_CONDB_PROXY"),
                "https": os.environ.get("BELLE2_CONDB_PROXY"),
            }
        # test the given url or try the known defaults
        base_url_list = ConditionsDB.get_base_urls(base_url)
 
        for url in base_url_list:
            
            self._base_url = url.rstrip("/") + "/"
            try:
                req = self._session.request("HEAD", self._base_url + "v2/globalTags")
                req.raise_for_status()
            except requests.RequestException as e:
                B2WARNING(f"Problem connecting to {url}:\n     {e}\n Trying next server ...")
            else:
                break
        else:
            B2FATAL("No working database servers configured, giving up")
 
        # We have a working server so change the api to return json instead of
        # xml, much easier in python, also request non-cached replies. We do
        # this now because for the server check above we're fine with cached
        # results.
        self._session.headers.update({"Accept": "application/json", "Cache-Control": "no-cache"})
 

Member Function Documentation

check_payloads()

def check_payloads	(		self,
			payloads,
			information = "payloadId"
	)

Check for the existence of payloads in the database.

Arguments:
    payloads (list((str,str))): A list of payloads to check for. Each
       payload needs to be a tuple of the name of the payload and the
       md5 checksum of the payload file.
    information (str): The information to be extracted from the
       payload dictionary

Returns:
    A dictionary with the payload identifiers (name, checksum) as keys
    and the requested information as values for all payloads which are already
    present in the database.

Definition at line 528 of file __init__.py.

    def check_payloads(self, payloads, information="payloadId"):
        """
        Check for the existence of payloads in the database.
 
        Arguments:
            payloads (list((str,str))): A list of payloads to check for. Each
               payload needs to be a tuple of the name of the payload and the
               md5 checksum of the payload file.
            information (str): The information to be extracted from the
               payload dictionary
 
        Returns:
            A dictionary with the payload identifiers (name, checksum) as keys
            and the requested information as values for all payloads which are already
            present in the database.
        """
 
        search_query = [{"name": e[0], "checksum": e[1]} for e in payloads]
        try:
            req = self.request("POST", "/checkPayloads", json=search_query)
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Cannot check for existing payloads: {e}")
            return {}
 
        result = {}
        for payload in req.json():
            module = payload["basf2Module"]["name"]
            checksum = payload["checksum"]
            result[(module, checksum)] = payload[information]
 
        return result
 

create_globalTag()

def create_globalTag	(		self,
			name,
			description,
			user
	)

Create a new globaltag

Definition at line 431 of file __init__.py.

    def create_globalTag(self, name, description, user):
        """
        Create a new globaltag
        """
        info = {"name": name, "description": description, "modifiedBy": user, "isDefault": False}
        try:
            req = self.request("POST", "/globalTag/DEV", f"Creating globaltag {name}", json=info)
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not create globaltag {name}: {e}")
            return None
 
        return req.json()
 

create_iov()

def create_iov	(		self,
			globalTagId,
			payloadId,
			firstExp,
			firstRun,
			finalExp,
			finalRun
	)

Create an iov.

Args:
    globalTagId (int): id of the globaltag, obtain with get_globalTagId()
    payloadId (int): id of the payload, obtain from create_payload() or get_payloads()
    firstExp (int): first experiment for which this iov is valid
    firstRun (int): first run for which this iov is valid
    finalExp (int): final experiment for which this iov is valid
    finalRun (int): final run for which this iov is valid

Returns:
    payloadIovId of the created iov, None if creation was not successful

Definition at line 622 of file __init__.py.

    def create_iov(self, globalTagId, payloadId, firstExp, firstRun, finalExp, finalRun):
        """
        Create an iov.
 
        Args:
            globalTagId (int): id of the globaltag, obtain with get_globalTagId()
            payloadId (int): id of the payload, obtain from create_payload() or get_payloads()
            firstExp (int): first experiment for which this iov is valid
            firstRun (int): first run for which this iov is valid
            finalExp (int): final experiment for which this iov is valid
            finalRun (int): final run for which this iov is valid
 
        Returns:
            payloadIovId of the created iov, None if creation was not successful
        """
        try:
            # try to convert all arguments except self to integers to make sure they are
            # valid.
            local_variables = locals()
            variables = {e: int(local_variables[e]) for e in
                         ["globalTagId", "payloadId", "firstExp", "firstRun", "finalExp", "finalRun"]}
        except ValueError:
            B2ERROR("create_iov: All parameters need to be integers")
            return None
 
        # try to create the iov
        try:
            req = self.request("POST", "/globalTagPayload/{globalTagId},{payloadId}"
                               "/payloadIov/{firstExp},{firstRun},{finalExp},{finalRun}".format(**variables))
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not create IOV: {e}")
            return None
 
        return req.json()["payloadIovId"]
 

create_payload()

def create_payload	(		self,
			module,
			filename,
			checksum = None
	)

Create a new payload

Args:
    module (str): name of the module
    filename (str): name of the file
    checksum (str): md5 hexdigest of the file. Will be calculated automatically if not given

Definition at line 581 of file __init__.py.

    def create_payload(self, module, filename, checksum=None):
        """
        Create a new payload
 
        Args:
            module (str): name of the module
            filename (str): name of the file
            checksum (str): md5 hexdigest of the file. Will be calculated automatically if not given
        """
        if checksum is None:
            checksum = file_checksum(filename)
 
        # this is the only complicated request as we have to provide a
        # multipart/mixed request which is not directly provided by the request
        # library.
        files = [
            (filename, open(filename, "rb").read(), "application/x-root"),
            ("json", json.dumps({"checksum": checksum, "isDefault": False}), "application/json"),
        ]
        # ok we have the two "files" we want to send, create multipart/mixed
        # body
        fields = []
        for name, contents, mimetype in files:
            rf = RequestField(name=name, data=contents)
            rf.make_multipart(content_type=mimetype)
            fields.append(rf)
 
        post_body, content_type = encode_multipart_formdata(fields)
        content_type = ''.join(('multipart/mixed',) + content_type.partition(';')[1:])
        headers = {'Content-Type': content_type}
 
        # now make the request. Note to self: if multipart/form-data would be
        # accepted this would be so much nicer here. but it works.
        try:
            req = self.request("POST", f"/package/dbstore/module/{encode_name(module)}/payload", data=post_body, headers=headers)
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not create Payload: {e}")
            return None
 
        return req.json()["payloadId"]
 

delete_iov()

def delete_iov	(		self,
			iovId
	)

Delete an iov

Args:
    iovId (int): id of the iov to be deleted

Definition at line 657 of file __init__.py.

    def delete_iov(self, iovId):
        """Delete an iov
 
        Args:
            iovId (int): id of the iov to be deleted
        """
        try:
            self.request("DELETE", f"/payloadIov/{iovId}")
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not delete IOV: {e}")
 

get_all_iovs()

def get_all_iovs	(		self,
			globalTag,
			exp = None,
			run = None,
			message = None,
			run_range = None,
			fully_contained = False
	)

Return list of all payloads in the given globaltag where each element is
a `PayloadInformation` instance

Parameters:
    gobalTag (str): name of the globaltag
    exp (int): if given limit the list of payloads to the ones valid for
        the given exp,run combination
    run (int): if given limit the list of payloads to the ones valid for
        the given exp,run combination
    message (str): additional message to show when downloading the
        payload information. Will be directly appended to
        "Obtaining lists of iovs for globaltag {globalTag}"
    run_range (tuple): if given limit the list of payloads to the ones
        overlapping with the given run range, if
    fully_contained (bool): if True and the run_range is not None it limits
        the list of payloads to the ones fully contained in the given run range

Warning:
    Both, exp and run, need to be given at the same time. Just supplying
    an experiment or a run number will not work

Definition at line 444 of file __init__.py.

    def get_all_iovs(self, globalTag, exp=None, run=None, message=None, run_range=None, fully_contained=False):
        """
        Return list of all payloads in the given globaltag where each element is
        a `PayloadInformation` instance
 
        Parameters:
            gobalTag (str): name of the globaltag
            exp (int): if given limit the list of payloads to the ones valid for
                the given exp,run combination
            run (int): if given limit the list of payloads to the ones valid for
                the given exp,run combination
            message (str): additional message to show when downloading the
                payload information. Will be directly appended to
                "Obtaining lists of iovs for globaltag {globalTag}"
            run_range (tuple): if given limit the list of payloads to the ones
                overlapping with the given run range, if
            fully_contained (bool): if True and the run_range is not None it limits
                the list of payloads to the ones fully contained in the given run range
 
        Warning:
            Both, exp and run, need to be given at the same time. Just supplying
            an experiment or a run number will not work
        """
        globalTag = encode_name(globalTag)
        if message is None:
            message = ""
        if run_range is not None:
            if fully_contained:
                message += f" [fully contained in {tuple(run_range)}]"
            else:
                message += f" [valid in {tuple(run_range)}]"
            run_range = IntervalOfValidity(run_range)
 
        if exp is not None:
            msg = f"Obtaining list of iovs for globaltag {globalTag}, exp={exp}, run={run}{message}"
            req = self.request("GET", "/iovPayloads", msg, params={'gtName': globalTag, 'expNumber': exp, 'runNumber': run})
        else:
            msg = f"Obtaining list of iovs for globaltag {globalTag}{message}"
            req = self.request("GET", f"/globalTag/{globalTag}/globalTagPayloads", msg)
        all_iovs = []
        for item in req.json():
            payload = item["payload" if 'payload' in item else "payloadId"]
            if "payloadIov" in item:
                iovs = [item['payloadIov']]
            else:
                iovs = item['payloadIovs']
 
            for iov in iovs:
                if run_range is not None:
                    iov_ = IntervalOfValidity(iov['expStart'], iov['runStart'], iov['expEnd'], iov['runEnd'])
                    if fully_contained:
                        if not iov_ & run_range == iov_:
                            continue
                    else:
                        if iov_ & run_range is None:
                            continue
                all_iovs.append(PayloadInformation.from_json(payload, iov))
 
        all_iovs.sort()
        return all_iovs
 

get_base_urls()

def get_base_urls (   given_url )

static

Resolve the list of server urls. If a url is given just return it.
Otherwise return servers listed in BELLE2_CONDB_SERVERLIST or the
builtin defaults

Arguments:
    given_url (str): Explicit base_url. If this is not None it will be
       returned as is in a list of length 1

Returns:
    a list of urls to try for database connectivity

Definition at line 242 of file __init__.py.

    def get_base_urls(given_url):
        """Resolve the list of server urls. If a url is given just return it.
        Otherwise return servers listed in BELLE2_CONDB_SERVERLIST or the
        builtin defaults
 
        Arguments:
            given_url (str): Explicit base_url. If this is not None it will be
               returned as is in a list of length 1
 
        Returns:
            a list of urls to try for database connectivity
        """
 
        base_url_list = ConditionsDB.BASE_URLS[:]
        base_url_env = os.environ.get("BELLE2_CONDB_SERVERLIST", None)
        if given_url is not None:
            base_url_list = [given_url]
        elif base_url_env is not None:
            base_url_list = base_url_env.split()
            B2INFO("Getting Conditions Database servers from Environment:")
            for i, url in enumerate(base_url_list, 1):
                B2INFO(f"  {i}. {url}")
        # try to escalate to https for all given urls
        full_list = []
        for url in base_url_list:
            if url.startswith("http://"):
                full_list.append("https" + url[4:])
            # but keep the http in case of connection problems
            full_list.append(url)
        return full_list
 

get_globalTagInfo()

def get_globalTagInfo	(		self,
			name
	)

Get the id of the globaltag with the given name. Returns either the
id or None if the tag was not found

Definition at line 400 of file __init__.py.

    def get_globalTagInfo(self, name):
        """Get the id of the globaltag with the given name. Returns either the
        id or None if the tag was not found"""
 
        try:
            req = self.request("GET", f"/globalTag/{encode_name(name)}")
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Cannot find globaltag '{name}': {e}")
            return None
 
        return req.json()
 

get_globalTags()

def get_globalTags

(

self

)

Get a list of all globaltags. Returns a dictionary with the globaltag
names and the corresponding ids in the database

Definition at line 374 of file __init__.py.

    def get_globalTags(self):
        """Get a list of all globaltags. Returns a dictionary with the globaltag
        names and the corresponding ids in the database"""
 
        try:
            req = self.request("GET", "/globalTags")
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not get the list of globaltags: {e}")
            return None
 
        result = {}
        for tag in req.json():
            result[tag["name"]] = tag
 
        return result
 

get_globalTagType()

def get_globalTagType	(		self,
			name
	)

Get the dictionary describing the given globaltag type (currently
one of DEV or RELEASE). Returns None if tag type was not found.

Definition at line 412 of file __init__.py.

    def get_globalTagType(self, name):
        """
        Get the dictionary describing the given globaltag type (currently
        one of DEV or RELEASE). Returns None if tag type was not found.
        """
        try:
            req = self.request("GET", "/globalTagType")
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not get list of valid globaltag types: {e}")
            return None
 
        types = {e["name"]: e for e in req.json()}
 
        if name in types:
            return types[name]
 
        B2ERROR(f"Unknown globaltag type: '{name}', please use one of {', '.join(types)}")
        return None
 

get_iovs()

def get_iovs	(		self,
			globalTagName,
			payloadName = None
	)

Return existing iovs for a given tag name. It returns a dictionary
which maps (payloadId, first runId, final runId) to iovId

Parameters:
  globalTagName(str): Global tag name.
  payloadName(str):   Payload name (if None, selection by name is
                      not performed.

Definition at line 684 of file __init__.py.

    def get_iovs(self, globalTagName, payloadName=None):
        """Return existing iovs for a given tag name. It returns a dictionary
        which maps (payloadId, first runId, final runId) to iovId
 
        Parameters:
          globalTagName(str): Global tag name.
          payloadName(str):   Payload name (if None, selection by name is
                              not performed.
        """
 
        try:
            req = self.request("GET", f"/globalTag/{encode_name(globalTagName)}/globalTagPayloads")
        except ConditionsDB.RequestError:
            # there could be just no iovs so no error
            return {}
 
        result = {}
        for payload in req.json():
            payloadId = payload["payloadId"]["payloadId"]
            if payloadName is not None:
                if payload["payloadId"]["basf2Module"]["name"] != payloadName:
                    continue
            for iov in payload["payloadIovs"]:
                iovId = iov["payloadIovId"]
                firstExp, firstRun = iov["expStart"], iov["runStart"]
                finalExp, finalRun = iov["expEnd"], iov["runEnd"]
                result[(payloadId, firstExp, firstRun, finalExp, finalRun)] = iovId
 
        return result
 

get_payloads()

def get_payloads	(		self,
			global_tag = None
	)

Get a list of all defined payloads (for the given global_tag or by default for all).
Returns a dictionary which maps (module, checksum) to the payload id.

Definition at line 505 of file __init__.py.

    def get_payloads(self, global_tag=None):
        """
        Get a list of all defined payloads (for the given global_tag or by default for all).
        Returns a dictionary which maps (module, checksum) to the payload id.
        """
 
        try:
            if global_tag:
                req = self.request("GET", f"/globalTag/{encode_name(global_tag)}/payloads")
            else:
                req = self.request("GET", "/payloads")
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Cannot get list of payloads: {e}")
            return {}
 
        result = {}
        for payload in req.json():
            module = payload["basf2Module"]["name"]
            checksum = payload["checksum"]
            result[(module, checksum)] = payload["payloadId"]
 
        return result
 

get_revisions()

def get_revisions	(		self,
			entries
	)

Get the revision numbers of payloads in the database.

Arguments:
    entries (list): A list of payload entries.
       Each entry must have the attributes module and checksum.

Returns:
    True if successful.

Definition at line 560 of file __init__.py.

    def get_revisions(self, entries):
        """
        Get the revision numbers of payloads in the database.
 
        Arguments:
            entries (list): A list of payload entries.
               Each entry must have the attributes module and checksum.
 
        Returns:
            True if successful.
        """
 
        result = self.check_payloads([(entry.module, entry.checksum) for entry in entries], "revision")
        if not result:
            return False
 
        for entry in entries:
            entry.revision = result.get((entry.module, entry.checksum), 0)
 
        return True
 

has_globalTag()

def has_globalTag	(		self,
			name
	)

Check whether the globaltag with the given name exists.

Definition at line 390 of file __init__.py.

    def has_globalTag(self, name):
        """Check whether the globaltag with the given name exists."""
 
        try:
            self.request("GET", f"/globalTag/{encode_name(name)}")
        except ConditionsDB.RequestError:
            return False
 
        return True
 

modify_iov()

def modify_iov	(		self,
			iovId,
			firstExp,
			firstRun,
			finalExp,
			finalRun
	)

Modify the validity range of a given iov

Args:
    iovId (int): id of the iov to be modified
    firstExp (int): first experiment for which this iov is valid
    firstRun (int): first run for which this iov is valid
    finalExp (int): final experiment for which this iov is valid
    finalRun (int): final run for which this iov is valid

Definition at line 668 of file __init__.py.

    def modify_iov(self, iovId, firstExp, firstRun, finalExp, finalRun):
        """Modify the validity range of a given iov
 
        Args:
            iovId (int): id of the iov to be modified
            firstExp (int): first experiment for which this iov is valid
            firstRun (int): first run for which this iov is valid
            finalExp (int): final experiment for which this iov is valid
            finalRun (int): final run for which this iov is valid
        """
        try:
            querystring = {"expStart": str(firstExp), "runStart": str(firstRun), "expEnd": str(finalExp), "runEnd": str(finalRun)}
            self.request("PUT", f"/payloadIov/{iovId}", params=querystring)
        except ConditionsDB.RequestError as e:
            B2ERROR(f"Could not modify IOV: {e}")
 

request()

def request	(		self,
			method,
			url,
			message = None,
		*	args,
		**	argk
	)

Request function, similar to requests.request but adding the base_url

Args:
    method (str): GET, POST, etc.
    url (str): url for the request, base_url will be prepended
    message (str): message to show when starting the request and if it fails

All other arguments will be forwarded to requests.request.

Definition at line 330 of file __init__.py.

    def request(self, method, url, message=None, *args, **argk):
        """
        Request function, similar to requests.request but adding the base_url
 
        Args:
            method (str): GET, POST, etc.
            url (str): url for the request, base_url will be prepended
            message (str): message to show when starting the request and if it fails
 
        All other arguments will be forwarded to requests.request.
        """
        if message is not None:
            B2INFO(message)
 
        try:
            req = self._session.request(method, self._base_url + "v2/" + url.lstrip("/"), *args, **argk)
        except requests.exceptions.ConnectionError as e:
            B2FATAL("Could not access '" + self._base_url + url.lstrip("/") + "': " + str(e))
 
        if req.status_code >= 300:
            # Apparently something is not good. Let's try to decode the json
            # reply containing reason and message
            try:
                response = req.json()
                message = response.get("message", "")
                colon = ": " if message.strip() else ""
                error = f"Request {method} {url} returned {response['code']} {response['reason']}{colon}{message}"
            except json.JSONDecodeError:
                # seems the reply was not even json
                error = f"Request {method} {url} returned non JSON response {req.status_code}: {req.content}"
 
            if message is not None:
                raise ConditionsDB.RequestError(f"{message} failed: {error}")
            else:
                raise ConditionsDB.RequestError(error)
 
        if method != "HEAD" and req.status_code != requests.codes.no_content:
            try:
                req.json()
            except json.JSONDecodeError as e:
                B2INFO(f"Invalid response: {req.content}")
                raise ConditionsDB.RequestError(f"{method} {url} returned invalid JSON response {e}")
        return req
 

set_authentication_token()

def set_authentication_token	(		self,
			token
	)

Set authentication token when talking to the database

Args:
    token (str): JWT to hand to the database. Will not be checked
        for validity here.

Definition at line 320 of file __init__.py.

    def set_authentication_token(self, token):
        """
        Set authentication token when talking to the database
 
        Args:
            token (str): JWT to hand to the database. Will not be checked
                for validity here.
        """
        self._session.auth = BearerAuth(token)
 

staging_request()

def staging_request	(		self,
			filename,
			normalize,
			data,
			password
	)

Upload a testing payload storage to a staging globaltag and create or update a jira issue

Parameters:
  filename (str): filename of the testing payload storage file that should be uploaded
  normalize (Union[bool, str]): if True the payload root files will be
    normalized to have the same checksum for the same content, if
    normalize is a string in addition the file name in the root file
    metadata will be set to it
  data (dict): a dictionary with the information provided by the user:

    * task: category of globaltag, either main, online, prompt, data, mc, or analysis
    * tag: the globaltag name
    * request: type of request, either Update, New, or Modification. The latter two imply task == main because
      if new payload classes are introduced or payload classes are modified then they will first be included in
      the main globaltag. Here a synchronization of code and payload changes has to be managed.
      If new or modified payload classes should be included in other globaltags they must already be in a release.
    * pull-request: number of the pull request containing new or modified payload classes,
      only for request == New or Modified
    * backward-compatibility: description of what happens if the old payload is encountered by the updated code,
      only for request == Modified
    * forward-compatibility: description of what happens if a new payload is encountered by the existing code,
      only for request == Modified
    * release: the required release version
    * reason: the reason for the request
    * description: a detailed description for the globaltag manager
    * issue: identifier of an existing jira issue (optional)
    * user: name of the user
    * time: time stamp of the request

  password: the password for access to jira or the access token and secret for oauth access

Returns:
  True if the upload and jira issue creation/upload was successful

Definition at line 859 of file __init__.py.

    def staging_request(self, filename, normalize, data, password):
        """
        Upload a testing payload storage to a staging globaltag and create or update a jira issue
 
        Parameters:
          filename (str): filename of the testing payload storage file that should be uploaded
          normalize (Union[bool, str]): if True the payload root files will be
            normalized to have the same checksum for the same content, if
            normalize is a string in addition the file name in the root file
            metadata will be set to it
          data (dict): a dictionary with the information provided by the user:
 
            * task: category of globaltag, either main, online, prompt, data, mc, or analysis
            * tag: the globaltag name
            * request: type of request, either Update, New, or Modification. The latter two imply task == main because
              if new payload classes are introduced or payload classes are modified then they will first be included in
              the main globaltag. Here a synchronization of code and payload changes has to be managed.
              If new or modified payload classes should be included in other globaltags they must already be in a release.
            * pull-request: number of the pull request containing new or modified payload classes,
              only for request == New or Modified
            * backward-compatibility: description of what happens if the old payload is encountered by the updated code,
              only for request == Modified
            * forward-compatibility: description of what happens if a new payload is encountered by the existing code,
              only for request == Modified
            * release: the required release version
            * reason: the reason for the request
            * description: a detailed description for the globaltag manager
            * issue: identifier of an existing jira issue (optional)
            * user: name of the user
            * time: time stamp of the request
 
          password: the password for access to jira or the access token and secret for oauth access
 
        Returns:
          True if the upload and jira issue creation/upload was successful
        """
 
        # determine the staging globaltag name
        data['tag'] = upload_global_tag(data['task'])
        if data['tag'] is None:
            data['tag'] = f"temp_{data['task']}_{data['user']}_{data['time']}"
 
        # create the staging globaltag if it does not exists yet
        if not self.has_globalTag(data['tag']):
            if not self.create_globalTag(data['tag'], data['reason'], data['user']):
                return False
 
        # upload the payloads
        B2INFO(f"Uploading testing database {filename} to globaltag {data['tag']}")
        entries = []
        if not self.upload(filename, data['tag'], normalize, uploaded_entries=entries):
            return False
 
        # get the dictionary for the jira issue creation/update
        if data['issue']:
            issue = data['issue']
        else:
            issue = jira_global_tag_v2(data['task'])
        if issue is None:
            issue = {"components": [{"name": "globaltag"}]}
 
        # create jira issue text from provided information
        if type(issue) is tuple:
            description = issue[1].format(**data)
            issue = issue[0]
        else:
            description = f"""
|*Upload globaltag*      | {data['tag']} |
|*Request reason*        | {data['reason']} |
|*Required release*      | {data['release']} |
|*Type of request*       | {data['request']} |
"""
            if 'pull-request' in data.keys():
                description += f"|*Pull request* | \\#{data['pull-request']} |\n"
            if 'backward-compatibility' in data.keys():
                description += f"|*Backward compatibility* | \\#{data['backward-compatibility']} |\n"
            if 'forward-compatibility' in data.keys():
                description += f"|*Forward compatibility* | \\#{data['forward-compatibility']} |\n"
            description += '|*Details* |' + ''.join(data['details']) + ' |\n'
            if data['task'] == 'online':
                description += '|*Impact on data taking*|' + ''.join(data['data_taking']) + ' |\n'
 
        # add information about uploaded payloads/IoVs
        description += '\nPayloads\n||Name||Revision||IoV||\n'
        for entry in entries:
            description += f"|{entry.module} | {entry.revision} | ({entry.iov_str()}) |\n"
 
        # create a new issue
        if type(issue) is dict:
            issue["description"] = description
            if "summary" in issue.keys():
                issue["summary"] = issue["summary"].format(**data)
            else:
                issue["summary"] = f"Globaltag request for {data['task']} by {data['user']} at {data['time']}"
            if "project" not in issue.keys():
                issue["project"] = {"key": "BII"}
            if "issuetype" not in issue.keys():
                issue["issuetype"] = {"name": "Task"}
            if data["task"] == "main":
                issue["labels"] = ["TUPPR"]
 
            B2INFO(f"Creating jira issue for {data['task']} globaltag request")
            if isinstance(password, str):
                response = requests.post('https://agira.desy.de/rest/api/latest/issue', auth=(data['user'], password),
                                         json={'fields': issue})
            else:
                fields = {'issue': json.dumps(issue)}
                if 'user' in data.keys():
                    fields['user'] = data['user']
                if password:
                    fields['token'] = password[0]
                    fields['secret'] = password[1]
                response = requests.post('https://b2-master.belle2.org/cgi-bin/jira_issue.py', data=fields)
            if response.status_code in range(200, 210):
                B2INFO(f"Issue successfully created: https://agira.desy.de/browse/{response.json()['key']}")
            else:
                B2ERROR('The creation of the issue failed: ' + requests.status_codes._codes[response.status_code][0])
                return False
 
        # comment on an existing issue
        else:
            # Let's make sure all assignees of new issues are added as watchers
            # in that case, otherwise they might never find out
            new_issue_config = jira_global_tag_v2(data['task'])
            if isinstance(new_issue_config, dict) and "assignee" in new_issue_config:
                user = new_issue_config['assignee'].get('name', None)
                if user is not None and isinstance(password, str):
                    response = requests.post(f'https://agira.desy.de/rest/api/latest/issue/{issue}/watchers',
                                             auth=(data['user'], password), json=user)
                    if response.status_code in range(200, 210):
                        B2INFO(f"Added {user} as watcher to {issue}")
                    else:
                        B2WARNING(f"Could not add {user} as watcher to {issue}: {response.status_code}")
 
            B2INFO(f"Commenting on jira issue {issue} for {data['task']} globaltag request")
            if isinstance(password, str):
                response = requests.post(f'https://agira.desy.de/rest/api/latest/issue/{issue}/comment',
                                         auth=(data['user'], password), json={'body': description})
            else:
                fields = {'id': issue, 'user': user, 'comment': description}
                if password:
                    fields['token'] = password[0]
                    fields['secret'] = password[1]
                response = requests.post('https://b2-master.belle2.org/cgi-bin/jira_issue.py', data=fields)
            if response.status_code in range(200, 210):
                B2INFO(f"Issue successfully updated: https://agira.desy.de/browse/{issue}")
            else:
                B2ERROR('The commenting of the issue failed: ' + requests.status_codes._codes[response.status_code][0])
                return False
 
        return True
 
 

upload()

def upload	(		self,
			filename,
			global_tag,
			normalize = False,
			ignore_existing = False,
			nprocess = 1,
			uploaded_entries = None
	)

Upload a testing payload storage to the conditions database.

Parameters:
  filename (str): filename of the testing payload storage file that should be uploaded
  global_tage (str): name of the globaltag to which the data should be uploaded
  normalize (Union[bool, str]): if True the payload root files will be normalized to have the same checksum for the
    same content, if normalize is a string in addition the file name in the root file metadata will be set to it
  ignore_existing (bool): if True do not upload payloads that already exist
  nprocess (int): maximal number of parallel uploads
  uploaded_entries (list): the list of successfully uploaded entries

Returns:
  True if the upload was successful

Definition at line 714 of file __init__.py.

    def upload(self, filename, global_tag, normalize=False, ignore_existing=False, nprocess=1, uploaded_entries=None):
        """
        Upload a testing payload storage to the conditions database.
 
        Parameters:
          filename (str): filename of the testing payload storage file that should be uploaded
          global_tage (str): name of the globaltag to which the data should be uploaded
          normalize (Union[bool, str]): if True the payload root files will be normalized to have the same checksum for the
            same content, if normalize is a string in addition the file name in the root file metadata will be set to it
          ignore_existing (bool): if True do not upload payloads that already exist
          nprocess (int): maximal number of parallel uploads
          uploaded_entries (list): the list of successfully uploaded entries
 
        Returns:
          True if the upload was successful
        """
 
        # first create a list of payloads
        from conditions_db.testing_payloads import parse_testing_payloads_file
        B2INFO(f"Reading payload list from {filename}")
        entries = parse_testing_payloads_file(filename)
        if entries is None:
            B2ERROR(f"Problems with testing payload storage file {filename}, exiting")
            return False
 
        if not entries:
            B2INFO(f"No payloads found in {filename}, exiting")
            return True
 
        B2INFO(f"Found {len(entries)} iovs to upload")
 
        # time to get the id for the globaltag
        tagId = self.get_globalTagInfo(global_tag)
        if tagId is None:
            return False
        tagId = tagId["globalTagId"]
 
        # now we could have more than one payload with the same iov so let's go over
        # it again and remove duplicates but keep the last one for each
        entries = sorted(set(reversed(entries)))
 
        if normalize:
            name = normalize if normalize is not True else None
            for e in entries:
                e.normalize(name=name)
 
        # so let's have a list of all payloads (name, checksum) as some payloads
        # might have multiple iovs. Each payload gets a list of all of those
        payloads = defaultdict(list)
        for e in entries:
            payloads[(e.module, e.checksum)].append(e)
 
        existing_payloads = {}
        existing_iovs = {}
 
        def upload_payload(item):
            """Upload a payload file if necessary but first check list of existing payloads"""
            key, entries = item
            if key in existing_payloads:
                B2INFO(f"{key[0]} (md5:{key[1]}) already existing in database, skipping.")
                payload_id = existing_payloads[key]
            else:
                entry = entries[0]
                payload_id = self.create_payload(entry.module, entry.filename, entry.checksum)
                if payload_id is None:
                    return False
 
                B2INFO(f"Created new payload {payload_id} for {entry.module} (md5:{entry.checksum})")
 
            for entry in entries:
                entry.payload = payload_id
 
            return True
 
        def create_iov(entry):
            """Create an iov if necessary but first check the list of existing iovs"""
            if entry.payload is None:
                return None
 
            iov_key = (entry.payload,) + entry.iov_tuple
            if iov_key in existing_iovs:
                entry.iov = existing_iovs[iov_key]
                B2INFO(f"IoV {entry.iov_tuple} for {entry.module} (md5:{entry.checksum}) already existing in database, skipping.")
            else:
                entry.payloadIovId = self.create_iov(tagId, entry.payload, *entry.iov_tuple)
                if entry.payloadIovId is None:
                    return False
 
                B2INFO(f"Created IoV {entry.iov_tuple} for {entry.module} (md5:{entry.checksum})")
 
            return entry
 
        # multithreading for the win ...
        with ThreadPoolExecutor(max_workers=nprocess) as pool:
            # if we want to check for existing payloads/iovs we schedule the download of
            # the full payload list. And write a message as each completes
            if not ignore_existing:
                B2INFO("Downloading information about existing payloads and iovs...")
                futures = []
                existing_iovs = {}
                existing_payloads = {}
 
                def create_future(iter, func, callback=None):
                    fn = pool.submit(iter, func)
                    if callback is not None:
                        fn.add_done_callback(callback)
                    futures.append(fn)
 
                def update_iovs(iovs):
                    existing_iovs.update(iovs.result())
                    B2INFO(f"Found {len(existing_iovs)} existing iovs in {global_tag}")
 
                def update_payloads(payloads):
                    existing_payloads.update(payloads.result())
                    B2INFO(f"Found {len(existing_payloads)} existing payloads")
 
                create_future(self.get_iovs, global_tag, update_iovs)
                # checking existing payloads should not be done with too many at once
                for chunk in chunks(payloads.keys(), 1000):
                    create_future(self.check_payloads, chunk, update_payloads)
 
                futures_wait(futures)
 
            # upload payloads
            failed_payloads = sum(0 if result else 1 for result in pool.map(upload_payload, payloads.items()))
            if failed_payloads > 0:
                B2ERROR(f"{failed_payloads} payloads could not be uploaded")
 
            # create IoVs
            failed_iovs = 0
            for entry in pool.map(create_iov, entries):
                if entry:
                    if uploaded_entries is not None:
                        uploaded_entries.append(entry)
                else:
                    failed_iovs += 1
            if failed_iovs > 0:
                B2ERROR(f"{failed_iovs} IoVs could not be created")
 
        # update revision numbers
        if uploaded_entries is not None:
            self.get_revisions(uploaded_entries)
 
        return failed_payloads + failed_iovs == 0
 

Member Data Documentation

_base_url

_base_url

protected

base url to be prepended to all requests

Definition at line 303 of file __init__.py.

_session

_session

protected

session object to get keep-alive support and connection pooling

Definition at line 284 of file __init__.py.

BASE_URLS

list BASE_URLS = [conditions.default_metadata_provider_url]

static

base url to the conditions db to be used if no custom url is given

Definition at line 235 of file __init__.py.

---

The documentation for this class was generated from the following file:

• framework/scripts/conditions_db/__init__.py