How to use the rucio.db.sqla.session.read_session function in rucio

@read_session
def get_global_account_limit(account, rse_expression, session=None):
    """
    Returns the global account limit for the account on the rse expression.

    :param account:         Account to check the limit for.
    :param rse_expression:  RSE expression to check the limit for.
    :param session:         Database session in use.
    :return:                Limit in Bytes.
    """
    try:
        global_account_limit = session.query(models.AccountGlobalLimit).filter_by(account=account, rse_expression=rse_expression).one()
        if global_account_limit.bytes == -1:
            return float("inf")
        else:
            return global_account_limit.bytes
    except NoResultFound:

@read_session
def get_rses_with_attribute(key, session=None):
    """
    Return all RSEs with a certain attribute.

    :param key: The key for the attribute.
    :param session: The database session in use.

    :returns: List of rse dictionaries
    """
    rse_list = []

    query = session.query(models.RSE).\
        join(models.RSEAttrAssociation, models.RSE.id == models.RSEAttrAssociation.rse_id).\
        filter(models.RSE.deleted == False, models.RSEAttrAssociation.key == key).group_by(models.RSE)  # NOQA

    for row in query:

    @read_session
    def get_dids(self, session=None):
        query = """select t.jeditaskid, t.username, t.status, d.datasetname from ATLAS_PANDA.JEDI_TASKS t
        inner join ATLAS_PANDA.JEDI_DATASETS d
        on t.jeditaskid = d.jeditaskid
        where t.creationdate > SYS_EXTRACT_UTC(systimestamp) - 5/(24*60) and t.tasktype = 'anal' and t.prodsourcelabel = 'user'
        and d.type = 'input'
        order by d.jeditaskid asc"""

        tasks = session.execute(query)

        for t in tasks.fetchall():
            status = t[2]
            if status == 'running':
                continue
            tid = t[0]
            if tid < self.max_tid:

@read_session
def get_sources(request_id, rse_id=None, session=None):
    """
    Retrieve sources by its ID.

    :param request_id:  Request-ID as a 32 character hex string.
    :param rse_id:      RSE ID as a 32 character hex string.
    :param session:     Database session to use.
    :returns:           Sources as a dictionary.
    """

    try:
        if rse_id:
            tmp = session.query(models.Source).filter_by(request_id=request_id, rse_id=rse_id).all()
        else:
            tmp = session.query(models.Source).filter_by(request_id=request_id).all()

@read_session
def list_rse_attributes(rse_id, session=None):
    """
    List RSE attributes for a RSE.

    :param rse_id:  The RSE id.
    :param session: The database session in use.

    :returns: A dictionary with RSE attributes for a RSE.
    """
    rse_attrs = {}

    query = session.query(models.RSEAttrAssociation).filter_by(rse_id=rse_id)
    for attr in query:
        rse_attrs[attr.key] = attr.value
    return rse_attrs

@read_session
def query_request(request_id, transfertool='fts3', session=None):
    """
    Query the status of a request.

    :param request_id:    Request-ID as a 32 character hex string.
    :param transfertool:  Transfertool name as a string.
    :param session:       Database session to use.
    :returns:             Request status information as a dictionary.
    """

    record_counter('core.request.query_request')

    req = get_request(request_id, session=session)

    req_status = {'request_id': request_id,
                  'new_state': None}

@read_session
def list_payload_counts(executable, older_than=600, hash_executable=None, session=None):
    """
    Give the counts of number of threads per payload for a certain executable.

    :param executable: Executable name as a string, e.g., conveyor-submitter
    :param older_than: Removes specified heartbeats older than specified nr of seconds
    :param hash_executable: Hash of the executable.

    :returns: List of tuples
    """

    if not hash_executable:
        hash_executable = calc_hash(executable)
    query = session.query(Heartbeats.payload,
                          func.count(Heartbeats.payload))\
                   .filter(Heartbeats.executable == hash_executable)\

@read_session
def is_scope_owner(scope, account, session=None):
    """ check to see if account owns the scope.

    :param scope: the scope to check.
    :param account: the account to check.
    :param session: The database session in use.

    :returns: True or false
    """
    return True if session.query(models.Scope).filter_by(scope=scope, account=account).first() else False

@read_session
def get_requests_by_transfer(external_host, transfer_id, session=None):
    """
    Retrieve requests by its transfer ID.

    :param request_host:  Name of the external host.
    :param transfer_id:   External transfer job id as a string.
    :param session:       Database session to use.
    :returns:             List of Requests.
    """

    try:
        tmp = session.query(models.Request).filter_by(external_id=transfer_id).all()

        if tmp:
            result = []
            for t in tmp:

@read_session
def get_updated_dids(total_workers, worker_number, limit=10, session=None):
    """
    Get updated dids.

    :param total_workers:      Number of total workers.
    :param worker_number:      id of the executing worker.
    :param limit:              Maximum number of dids to return.
    :param session:            Database session in use.
    """
    query = session.query(models.UpdatedDID.id,
                          models.UpdatedDID.scope,
                          models.UpdatedDID.name,
                          models.UpdatedDID.rule_evaluation_action)

    if total_workers > 0:
        if session.bind.dialect.name == 'oracle':