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

@transactional_session
def truncate_messages(session=None):
    """
    Delete all stored messages. This is for internal purposes only.

    :param session: The database session to use.
    """

    try:
        session.query(Message).delete(synchronize_session=False)
    except IntegrityError as e:
        raise RucioException(e.args)

@transactional_session
def release_waiting_requests_fifo(rse_id, activity=None, count=None, account=None, direction='destination', session=None):
    """
    Release waiting requests. Transfer requests that were requested first, get released first (FIFO).

    :param rse_id:           The RSE id.
    :param activity:         The activity.
    :param count:            The count to be released.
    :param account:          The account name whose requests to release.
    :param direction:        Direction if requests are grouped by source RSE or destination RSE.
    :param session:          The database session.
    """

    dialect = session.bind.dialect.name
    rowcount = 0
    if dialect == 'mysql':
        subquery = session.query(models.Request.id)\

@transactional_session
def __resolve_bytes_length_events_did(scope, name, session):
    """
    Resolve bytes, length and events of a did

    :param scope:   The scope of the DID.
    :param name:    The name of the DID.
    :param session: The database session in use.
    """

    try:
        did = session.query(models.DataIdentifier).filter_by(scope=scope, name=name).one()
    except NoResultFound:
        raise exception.DataIdentifierNotFound("Data identifier '%s:%s' not found" % (scope, name))

    bytes, length, events = 0, 0, 0
    if did.did_type == DIDType.FILE:

@transactional_session
def update_subscription(name, account, metadata=None, session=None):
    """
    Updates a subscription

    :param name: Name of the subscription
    :type:  String
    :param account: Account identifier
    :type account:  String
    :param metadata: Dictionary of metadata to update. Supported keys : filter, replication_rules, comments, lifetime, retroactive, dry_run, priority, last_processed
    :type metadata:  Dict
    :param session: The database session in use.
    :raises: SubscriptionNotFound if subscription is not found
    """
    try:
        keep_history = get('subscriptions', 'keep_history')
    except ConfigNotFound:

@transactional_session
def __add_collections_to_container(scope, name, collections, account, session):
    """
    Add collections (datasets or containers) to container.

    :param scope: The scope name.
    :param name: The container name.
    :param collections: .
    :param account: The account owner.
    :param session: The database session in use.
    """

    condition = or_()
    for cond in collections:

        if (scope == cond['scope']) and (name == cond['name']):
            raise exception.UnsupportedOperation('Self-append is not valid!')

@transactional_session
def get_auth_token_ssh(account, signature, appid, ip=None, session=None):
    """
    Authenticate a Rucio account temporarily via SSH key exchange.

    The token lifetime is 1 hour.

    :param account: Account identifier as a string.
    :param signature: Response to server challenge signed with SSH private key as string.
    :param appid: The application identifier as a string.
    :param ip: IP address of the client as a string.
    :param session: The database session in use.

    :returns: Authentication token as a Python struct
              .token string
              .expired_at datetime
    """

@transactional_session
def add_account_attribute(account, key, value, session=None):
    """
    Add an attribute for the given account name.

    :param key: the key for the new attribute.
    :param value: the value for the new attribute.
    :param account: the account to add the attribute to.
    :param session: The database session in use.
    """

    query = session.query(models.Account).filter_by(account=account, status=AccountStatus.ACTIVE)

    try:
        query.one()
    except exc.NoResultFound:
        raise exception.AccountNotFound("Account ID '{0}' does not exist".format(account))

@transactional_session
def successful_transfer(scope, name, rse_id, nowait, session=None):
    """
    Update the state of all replica locks because of an successful transfer

    :param scope:    Scope of the did
    :param name:     Name of the did
    :param rse_id:   RSE id
    :param nowait:   Nowait parameter for the for_update queries.
    :param session:  DB Session.
    """

    locks = session.query(models.ReplicaLock).with_for_update(nowait=nowait).filter_by(scope=scope, name=name, rse_id=rse_id)
    for lock in locks:
        if lock.state == LockState.OK:
            continue
        logging.debug('Marking lock %s:%s for rule %s on rse %s as OK' % (lock.scope, lock.name, str(lock.rule_id), get_rse_name(rse_id=lock.rse_id, session=session)))

@transactional_session
def set_rse_transfer_limits(rse_id, activity, rse_expression=None, max_transfers=0, transfers=0, waitings=0, volume=0, session=None):
    """
    Set RSE transfer limits.

    :param rse_id: The RSE id.
    :param activity: The activity.
    :param rse_expression: RSE expression string.
    :param max_transfers: Maximum transfers.
    :param transfers: Current number of tranfers.
    :param waitings: Current number of waitings.
    :param volume: Maximum transfer volume in bytes.
    :param session: The database session in use.

    :returns: True if successful, otherwise false.
    """
    try:

@transactional_session
def update_vo(vo, parameters, session=None):
    """
    Update VO properties (email, description).

    :param vo: The VO to update.
    :param parameters: A dictionary with the new properties.
    :param session: The db session in use.
    """
    try:
        query = session.query(models.VO).filter_by(vo=vo).one()
    except NoResultFound:
        raise exception.VONotFound('VO {} not found'.format(vo))
    param = {}
    for key in parameters:
        if key in ['email', 'description']:
            param[key] = parameters[key]