o ge@sdZddlZddlZddlmZddlmZddlm Z ddl m Z m Z m Z mZmZmZmZddlZddlmZdd lmZmZmZmZmZmZGd d d ZGd d d ZdS)z2 Classes for interacting with Salesforce Bulk API N) OrderedDict)partial)sleep)AnyDictIterableListOptionalUnioncast)SalesforceGeneralError) BulkDataAny BulkDataStrHeadersProxiescall_salesforcelist_from_generatorc @sLeZdZdZ  ddededeedeejfddZ d ed d fd d Z dS) SFBulkHandlerz Bulk API request handler Intermediate class which allows us to use commands, such as 'sf.bulk.Contacts.create(...)' This is really just a middle layer, whose sole purpose is to allow the above syntax N session_idbulk_urlproxiessessioncCsB||_|pt|_||_|s|dur||j_d|jdd|_dS)aInitialize the instance with the given parameters. Arguments: * session_id -- the session ID for authenticating to Salesforce * bulk_url -- API endpoint set in Salesforce instance * proxies -- the optional map of scheme to proxy server * session -- Custom requests session, created in calling code. This enables the use of requests Session features not otherwise exposed by simple_salesforce. Nzapplication/json1)z Content-TypezX-SFDC-Sessionz X-PrettyPrint)rrequestsSessionrrrheaders)selfrrrrrO/home/ubuntu/webapp/venv/lib/python3.10/site-packages/simple_salesforce/bulk.py__init__s  zSFBulkHandler.__init__namereturn SFBulkTypecCst||j|j|jdS)N) object_namerrr)r#rrr)rr!rrr __getattr__;s zSFBulkHandler.__getattr__)NN) __name__ __module__ __qualname____doc__strr rrrr r%rrrrrs$  !rc@seZdZdZdedededejfddZ dd eded e d e ed*eeefd!ed"e d#e d eee fd+d,Z )   d?ded*ed e d"e d#e d ee f d-d.Z )   d?ded*ed e d"e d#e d ee f d/d0Z )   d?ded ed*ed e d"e d#e d ee fd1d2Z )   d?ded*ed e d"e d#e d ee f d3d4Z )   d?ded*ed e d"e d#e d ee f d5d6Z!  d@ded7e d!ed ee fd8d9Z"  d@ded7e d!ed ee fd:d;Z#dS)Ar#z& Interface to Bulk/Async API functionsr$rrrcCs||_||_||_||_dS)aInitialize the instance with the given parameters. Arguments: * object_name -- the name of the type of SObject this represents, e.g. `Lead` or `Contact` * bulk_url -- API endpoint set in Salesforce instance * headers -- bulk API headers * session -- Custom requests session, created in calling code. This enables the use of requests Session features not otherwise exposed by simple_salesforce. N)r$rrr)rr$rrrrrrr Hs zSFBulkType.__init__N operation use_serialexternal_id_fieldr"c Cs`||j|rdnddd}|dkr||d<|jd}t|d|j|jtj|d d d }|jtd S) z Create a bulk job Arguments: * operation -- Bulk operation to be performed by job * use_serial -- Process batches in order * external_id_field -- unique identifier field for upsert operations r rJSON)r+objectconcurrencyMode contentTypeupsertexternalIdFieldNamejobPOSTF allow_nanurlmethodrrdataobject_pairs_hook)r$rrrrjsondumpsr)rr+r,r-payloadr9resultrrr _create_job`s"   zSFBulkType._create_jobjob_idc CsDddi}|jd|}t|d|j|jtj|ddd}|jtdS) z Close a bulk job stateClosedjob/r5Fr6r8r<)rrrrr>r?r)rrCr@r9rArrr _close_jobs zSFBulkType._close_jobcCs0|jd|}t|d|j|jd}|jtdS)z) Get an existing job to check the status rFGETr9r:rrr<rrrrr>r)rrCr9rArrr_get_jobs zSFBulkType._get_jobr;cCsP|jd|d}|dvrtj|dd}n|}t|d|j|j|d}|jtdS) z Add a set of data as a batch to an existing job Separating this out in case of later implementations involving multiple batches rFz/batchqueryqueryAllFr6r5r8r<)rr>r?rrrr)rrCr;r+r9data_rArrr _add_batchs  zSFBulkType._add_batchbatch_idcCs6|jd|d|}t|d|j|jd}|jtdS)z+ Get an existing batch to check the status rF/batch/rHrIr<rJ)rrCrQr9rArrr _get_batchs zSFBulkType._get_batchc cs|jd|d|d}t|d|j|jd}|dvr:|D]}|d|}t|d|j|jd}|VqdS|VdS) 0 retrieve a set of results from a completed job rFrRz/resultrHrIrL/N)rrrrr>) rrCrQr+r9rA batch_resulturl_query_resultsbatch_query_resultrrr_get_batch_resultss(  zSFBulkType._get_batch_resultsc cs|jd|d|d}t|d|j|jd}|j||dd}g}t|D] \}}dd ||D} | D]} || q8| |q%|Vd S) rTrFrRz/requestrHrI batch_results)r+cSsPg|]$\}}t|tr"|dt|d|t|din||iqS).r) isinstancedictlistkeysget).0kvrrr s zDSFBulkType._get_batch_request_with_batch_results..N) rrrrrY enumerater>itemsupdateappend) rrCrQr9 batch_requestrVresultsidxiflattened_request_dict request_fieldrrr%_get_batch_request_with_batch_resultss(    z0SFBulkType._get_batch_request_with_batch_resultsFbatchwaitbypass_resultsinclude_detailed_resultscCs|sE|j|d|ddd}|dvr(t||j|d|ddd}|dvs|r7|j|d|dd}|S|j|d|d|d}|S||ddg}|S)z Gets batches from concurrent worker threads. self._bulk_operation passes batch jobs. The worker function checks each batch job waiting for it complete and appends the results. jobIdidrCrQrD CompletedFailed NotProcessedrCrQr+)rsrC)rSrrorY)rrqr+rrrsrt batch_statusrArrrworkers:    zSFBulkType.workerr4c sd}d}g}d}d\}} t|D]2\} } ttj| tdd} t| | |k||kgr:|||| | }d\}} | | 7} |d7}q|t|dkrT|||dfd d |DS) a Auto-create batches that respect bulk api V1 limits. bulk v1 api has following limits number of records <= 10000 AND file_size_limit <= 10MB AND number_of_character_limit <= 10000000 Documentation on limits can be found at: https://developer.salesforce.com/docs/atlas.en-us .salesforce_app_limits_cheatsheet.meta /salesforce_app_limits_cheatsheet /salesforce_app_limits_platform_bulkapi.htm#ingest_jobs Our JSON serialization uses the default `ensure_ascii=True`, so the character and byte lengths will be the same. Therefore we only need to adhere to a single length limit of 10,000,000 characters. TODO: In future when simple-salesforce supports bulk api V2 we should detect api version and set max file size accordingly. V2 increases file size limit to 150MB TODO: support for the following limits have not been added since these are record / field level limits and not chunk level limits: * Maximum number of fields in a record: 5,000 * Maximum number of characters in a record: 400,000 * Maximum number of characters in a field: 131,072 'ir)rr)defaultr Ncsg|] }j|dqS)rCr;r+rPrarlr4r+rrrrdxs  z5SFBulkType._add_autosized_batches..)relenr>r?r*anyrh) rr;r+r4 record_limit char_limitbatches last_break record_count char_countrlrecordadditional_charsrrr_add_autosized_batches:s2$  z!SFBulkType._add_autosized_batchesr batch_sizec stts dks tddvrstddkr&ttdtjh} j||ddkrCj dd} n t tfd d fd d t td DD} t j |||d } | | | } |s|dd | Dndd | D} jddWd| S1swY| Sdvrj||djdd}jddj|d|dd}|ddvrt|j|d|dd}|ddvs|ddkrtd|d|d|dj|d|dd} | S)a String together helper functions to create a complete end-to-end bulk API request Arguments: * operation -- Bulk operation to be performed by job * data -- list of dict to be passed as a batch * use_serial -- Process batches in serial mode * external_id_field -- unique identifier field for upsert operations * wait -- seconds to sleep between checking batch status * batch_size -- number of records to assign for each batch in the job or `auto` autoz'batch size should be auto or an integerrLzdata should not be empty for r)r+r,r-rv)r4r;r+cs$g|]}|rjd|dqS)rvrrrrrrrds z.SFBulkType._bulk_operation..cs$g|]}||dqS)r rr)rr;rrrdsr )r+rrrsrtcSs$g|]}|D] }|D]}|q qqSrr)rasublistrlxrrrrds cSs0g|]}|D]}|D]\}}||iq qqSr)rf)rarrlrbrcrrrrds )rCNrrurwrDrxrz stateMessager|)r\int ValueErrorminr concurrentfuturesThreadPoolExecutorrBrr rangerr~maprGrPrSrr rY)rr+r;r,r-rrrrsrtpoolrmulti_thread_workerlist_of_resultsrjrqr}r)rr;r4r+rr_bulk_operation~s     NN       zSFBulkType._bulk_operationcC|j|d||||d}|S)z soft delete records Data is batched by 10,000 records by default. To pick a lower size pass smaller integer to `batch_size`. to let simple-salesforce pick the appropriate limit dynamically, enter `batch_size='auto'` deleter,r+r;rrsrtrrr;rr,rsrtrjrrrrzSFBulkType.deletecCr)z insert records Data is batched by 10,000 records by default. To pick a lower size pass smaller integer to `batch_size`. to let simple-salesforce pick the appropriate limit dynamically, enter `batch_size='auto'` insertrrrrrrrrzSFBulkType.insertc Cs|j|d|||||d}|S)a upsert records based on a unique identifier Data is batched by 10,000 records by default. To pick a lower size pass smaller integer to `batch_size`. to let simple-salesforce pick the appropriate limit dynamically, enter `batch_size='auto'` r2)r,r+r-r;rrsrtr)rr;r-rr,rsrtrjrrrr2(s zSFBulkType.upsertcCr)z update records Data is batched by 10,000 records by default. To pick a lower size pass smaller integer to `batch_size`. to let simple-salesforce pick the appropriate limit dynamically, enter `batch_size='auto'` rgrrrrrrrgBrzSFBulkType.updatecCr)z hard delete records Data is batched by 10,000 records by default. To pick a lower size pass smaller integer to `batch_size`. to let simple-salesforce pick the appropriate limit dynamically, enter `batch_size='auto'` hardDeleterrrrrr hard_deleteZrzSFBulkType.hard_deletelazy_operationcC |jd||d}|r |St|S)z bulk query rMr+r;rrrrrr;rrrrjrrrrMrszSFBulkType.querycCr)z bulk queryAll rNrrrrrr query_allszSFBulkType.query_all)N)rpFF)FNrrpFF)rFFF)Frp)$r&r'r(r)r*rrrr boolr rrBrGrKrrPrSrrYrorrr~rrr rrrrr2rgrrMrrrrrr#Es  $       (  ( H   }      r#)r)concurrent.futuresrr> collectionsr functoolsrtimertypingrrrrr r r r exceptionsr utilrrrrrrrr#rrrrs   $  3