dragonchain/webserver/lib/transactions.py
def submit_bulk_transaction_v1(bulk_transaction: Sequence[Dict[str, Any]], api_key: "api_key_model.APIKeyModel") -> Dict[str, List[Any]]:
    """
    Formats, validates and enqueues the transactions in
    the payload of bulk_transaction
    Returns dictionary of 2 lists, key "201" are successfully created txn ids and key "400" are transactions that failed to post (entire dto passed in from user)
    """
    _log.info("[TRANSACTION_BULK] Checking if key is allowed to create all given bulk transactions")
    requested_types = set()
    for transaction in bulk_transaction:
        requested_types.add(transaction["txn_type"])
    # Check if allowed to create all these transactions of (potentially) various types
    if not api_key.is_key_allowed("transactions", "create", "create_transaction", False, extra_data={"requested_types": requested_types}):
        raise exceptions.ActionForbidden("API Key is not allowed to create all of the provided transaction types")

    _log.info(f"[TRANSACTION_BULK] Auth successful. Attempting to enqueue {len(bulk_transaction)} transactions")
    success = []
    fail = []
    pipeline = dc_redis.pipeline_sync()
    for transaction in bulk_transaction:
        try:
            txn_model = _generate_transaction_model(transaction)
            queue.enqueue_l1_pipeline(pipeline, txn_model.export_as_queue_task())
            success.append(txn_model.txn_id)
        except Exception:
            fail.append(transaction)
    # Queue the actual transactions in redis now
    for result in pipeline.execute():
        if not result:
            raise RuntimeError("Failed to enqueue")

    return {"201": success, "400": fail}

    Formats, validates and enqueues the transactions in
    the payload of bulk_transaction
    Returns dictionary of 2 lists, key "201" are successfully created txn ids and key "400" are transactions that failed to post (entire dto passed in from user)
    
Formats, validates and enqueues the transactions in
the payload of bulk_transaction
Returns dictionary of 2 lists, key "201" are successfully created txn ids and key "400" are transactions that failed to post (entire dto passed in from user)
Formats validates and enqueues the transactions in the payload of bulk_transaction Returns dictionary of 2 lists key " 201 " are successfully created txn ids and key " 400 " are transactions that failed to post ( entire dto passed in from user )
def submit_bulk_transaction_v1(bulk_transaction: Sequence[Dict[str, Any]], api_key: "api_key_model.APIKeyModel") -> Dict[str, List[Any]]:
    _log.info("[TRANSACTION_BULK] Checking if key is allowed to create all given bulk transactions")
    requested_types = set()
    for transaction in bulk_transaction:
        requested_types.add(transaction["txn_type"])
    if not api_key.is_key_allowed("transactions", "create", "create_transaction", False, extra_data={"requested_types": requested_types}):
        raise exceptions.ActionForbidden("API Key is not allowed to create all of the provided transaction types")
    _log.info(f"[TRANSACTION_BULK] Auth successful. Attempting to enqueue {len(bulk_transaction)} transactions")
    success = []
    fail = []
    pipeline = dc_redis.pipeline_sync()
    for transaction in bulk_transaction:
        try:
            txn_model = _generate_transaction_model(transaction)
            queue.enqueue_l1_pipeline(pipeline, txn_model.export_as_queue_task())
            success.append(txn_model.txn_id)
        except Exception:
            fail.append(transaction)
    for result in pipeline.execute():
        if not result:
            raise RuntimeError("Failed to enqueue")
    return {"201": success, "400": fail}
def submit_bulk_transaction_v1 ( bulk_transaction : Sequence [ Dict [ str , Any ] ] , api_key : "api_key_model.APIKeyModel" ) -> Dict [ str , List [ Any ] ] : _log . info ( "[TRANSACTION_BULK] Checking if key is allowed to create all given bulk transactions" ) requested_types = set ( ) for transaction in bulk_transaction : requested_types . add ( transaction [ "txn_type" ] ) if not api_key . is_key_allowed ( "transactions" , "create" , "create_transaction" , False , extra_data = { "requested_types" : requested_types } ) : raise exceptions . ActionForbidden ( "API Key is not allowed to create all of the provided transaction types" ) _log . info ( f"[TRANSACTION_BULK] Auth successful. Attempting to enqueue {len(bulk_transaction)} transactions" ) success = [ ] fail = [ ] pipeline = dc_redis . pipeline_sync ( ) for transaction in bulk_transaction : try : txn_model = _generate_transaction_model ( transaction ) queue . enqueue_l1_pipeline ( pipeline , txn_model . export_as_queue_task ( ) ) success . append ( txn_model . txn_id ) except Exception : fail . append ( transaction ) for result in pipeline . execute ( ) : if not result : raise RuntimeError ( "Failed to enqueue" ) return { "201" : success , "400" : fail }
Formats, validates and enqueues the transactions in
the payload of bulk_transaction
Returns dictionary of 2 lists, key "201" are successfully created txn ids and key "400" are transactions that failed to post (entire dto passed in from user)
Formats validates and enqueues the transactions in the payload of bulk_transaction Returns dictionary of 2 lists key " 201 " are successfully created txn ids and key " 400 " are transactions that failed to post ( entire dto passed in from user )
"""
    Formats, validates and enqueues the transactions in
    the payload of bulk_transaction
    Returns dictionary of 2 lists, key "201" are successfully created txn ids and key "400" are transactions that failed to post (entire dto passed in from user)
    """ # Check if allowed to create all these transactions of (potentially) various types # Queue the actual transactions in redis now
bulk_transaction
Sequence[Dict[str, Any]]
api_key
"api_key_model.APIKeyModel"