ɯeiNddlZddlmZddlmZddlmZmZmZm Z m Z m Z ddl Z ddlmZddlmZddlmZddlmZdd lmZdd lmZmZmZdd lmZdd lmZdd l m!Z!erddl"Z ddl#Z ee$Z%GddeZ&GddZ'y)N)Enum) getLogger) TYPE_CHECKINGIteratorListLiteralOptionalUnion)ASYNC_RETRY_PATTERN) DatabaseError)pandas)result_scan_statement)Query)is_in_stored_procedureresult_set_to_iterresult_set_to_rows)SnowparkSQLException)col)Rowc0eZdZdZdZdZdZdZdZdZ dZ d Z y ) _AsyncResultTyperow row_iteratorr pandas_batchescount no_resultupdatedeletemergeN) __name__ __module__ __qualname__ROWITERATORPANDAS PANDAS_BATCHCOUNT NO_RESULTUPDATEDELETEMERGE^/var/www/html/glpi_dashboard/venv/lib/python3.12/site-packages/snowflake/snowpark/async_job.pyrrs/ CH F#L EI F F Er-rc>eZdZdZej ddddfdedeeddd ed eee d e d e d ee ddfdZ e deefdZddZde fdZde fdZdefdZd dZdeeeeefd ededfdZ d!d eeddeeeeedede ddddf fdZy)"AsyncJoba Provides a way to track an asynchronous query in Snowflake. A :class:`DataFrame` object can be evaluated asynchronously and an :class:`AsyncJob` object will be returned. With this instance, you can: - retrieve results; - check the query status (still running or done); - cancel the running query; - retrieve the query ID and perform other operations on this query ID manually. :class:`AsyncJob` can be created by :meth:`Session.create_async_job` or action methods in :class:`DataFrame` and other classes. All methods in :class:`DataFrame` with a suffix of ``_nowait`` execute asynchronously and create an :class:`AsyncJob` instance. They are also equivalent to corresponding functions in :class:`DataFrame` and other classes that set ``block=False``. Therefore, to use it, you need to create a dataframe first. Here we demonstrate how to do that: First, we create a dataframe: >>> from snowflake.snowpark.functions import when_matched, when_not_matched >>> from snowflake.snowpark.types import IntegerType, StringType, StructField, StructType >>> df = session.create_dataframe([[float(4), 3, 5], [2.0, -4, 7], [3.0, 5, 6],[4.0,6,8]], schema=["a", "b", "c"]) Example 1 :meth:`DataFrame.collect` can be performed asynchronously:: >>> async_job = df.collect_nowait() >>> async_job.result() [Row(A=4.0, B=3, C=5), Row(A=2.0, B=-4, C=7), Row(A=3.0, B=5, C=6), Row(A=4.0, B=6, C=8)] You can also do:: >>> async_job = df.collect(block=False) >>> async_job.result() [Row(A=4.0, B=3, C=5), Row(A=2.0, B=-4, C=7), Row(A=3.0, B=5, C=6), Row(A=4.0, B=6, C=8)] Example 2 :meth:`DataFrame.to_pandas` can be performed asynchronously:: >>> async_job = df.to_pandas(block=False) >>> async_job.result() A B C 0 4.0 3 5 1 2.0 -4 7 2 3.0 5 6 3 4.0 6 8 Example 3 :meth:`DataFrame.first` can be performed asynchronously:: >>> async_job = df.first(block=False) >>> async_job.result() [Row(A=4.0, B=3, C=5)] Example 4 :meth:`DataFrame.count` can be performed asynchronously:: >>> async_job = df.count(block=False) >>> async_job.result() 4 Example 5 Save a dataframe to table or copy it into a stage file can also be performed asynchronously:: >>> table_name = "name" >>> async_job = df.write.save_as_table(table_name, block=False) >>> # copy into a stage file >>> remote_location = f"{session.get_session_stage()}/name.csv" >>> async_job = df.write.copy_into_location(remote_location, block=False) >>> async_job.result()[0]['rows_unloaded'] 4 Example 7 :meth:`Table.merge`, :meth:`Table.update`, :meth:`Table.delete` can also be performed asynchronously:: >>> schema = StructType([StructField("key", IntegerType()), StructField("value", StringType())]) >>> target_df = session.create_dataframe([(10, "old"), (10, "too_old"), (11, "old")], schema=schema) >>> target_df.write.save_as_table("my_table", mode="overwrite", table_type="temporary") >>> target = session.table("my_table") >>> source = session.create_dataframe([(10, "new"), (12, "new"), (13, "old")], schema=schema) >>> async_job = target.merge(source,target["key"] == source["key"],[when_matched().update({"value": source["value"]}),when_not_matched().insert({"key": source["key"]})],block=False) >>> async_job.result() MergeResult(rows_inserted=2, rows_updated=2, rows_deleted=0) Example 8 Cancel the running query associated with the dataframe:: >>> df = session.sql("select SYSTEM$WAIT(3)") >>> async_job = df.collect_nowait() >>> async_job.cancel() Example 9 Creating an :class:`AsyncJob` from an existing query ID, retrieving results and converting it back to a :class:`DataFrame`: >>> from snowflake.snowpark.functions import col >>> query_id = session.sql("select 1 as A, 2 as B, 3 as C").collect_nowait().query_id >>> async_job = session.create_async_job(query_id) >>> async_job.query # doctest: +SKIP 'select 1 as A, 2 as B, 3 as C' >>> async_job.result() [Row(A=1, B=2, C=3)] >>> async_job.result(result_type="pandas") A B C 0 1 2 3 >>> df = async_job.to_df() >>> df.select(col("A").as_("D"), "B").collect() [Row(D=1, B=2)] Example 10 Checking the status of a failed query (division by zero) using the new status APIs:: >>> import time >>> failing_query = session.sql("select 1/0 as result") >>> async_job = failing_query.collect_nowait() >>> while not async_job.is_done(): ... time.sleep(1.0) >>> async_job.is_done() True >>> async_job.is_failed() True >>> async_job.status() 'FAILED_WITH_ERROR' Note: - If a dataframe is associated with multiple queries: + if you use :meth:`Session.create_dataframe` to create a dataframe from a large amount of local data and evaluate this dataframe asynchronously, data will still be loaded into Snowflake synchronously, and only fetching data from Snowflake again will be performed asynchronously. + otherwise, multiple queries will be wrapped into a `Snowflake Anonymous Block `_ and executed asynchronously as one query. - Temporary objects (e.g., tables) might be created when evaluating dataframes and they will be dropped automatically after all queries finish when calling a synchronous API. When you evaluate dataframes asynchronously, temporary objects will only be dropped after calling :meth:`result`. NFTquery_idquerysessionz"snowflake.snowpark.session.Session result_type post_actionslog_on_exceptioncase_sensitivenum_statementsreturnc  "||_||_d|_||_|jjj |_||_|r|ng|_||_ ||_ ||_ | |_ d|_ d|_d|_d|_y)NTF)r1_query_can_query_be_retrieved_session_conncursor_cursor _result_type _post_actions_log_on_exception_case_sensitive_num_statements _parameters _result_meta _inserted_updated_deleted) selfr1r2r3r4r5r6r7r8kwargss r.__init__zAsyncJob.__init__s&  '+$ }}**113 '-9\r!1--!   r-cX|jsyd|j}|js |jj dj t d|jk(jdj}t|tsJt|dk(r tj|dd|_yt|dd|_|jS|jS#t$r+}tj|d |d|_Yd}~yd}~wwxYw) zs The SQL text of of the executed query. Returns ``None`` if it cannot be retrieved from Snowflake. Nz(query cannot be retrieved from query ID z information_schema.query_historyr1 query_textrz: result is emptyFz: )r<r1r;r=table_functionwhererselect'_internal_collect_with_tag_no_telemetry isinstancelistlen_loggerdebugstrr)rK error_messageresultexs r.r2zAsyncJob.querys ++Ft}}oVM;;8 445WXs:$--?@ -@@B &fd3336{a' 7H&IJ7<4#&)&)A,&7 ;; 4;; , MM]O2bT":;38D0 sAC55 D)>!D$$D)c^|jjt|jS)z^ Returns a :class:`DataFrame` built from the result of this asynchronous job. )r=sqlrr1)rKs r.to_dfzAsyncJob.to_dfs#}}  !6t}}!EFFr-c|jjjj|j}|jjjj |}| S)z Checks the status of the query associated with this instance and returns a bool value indicating whether the query has finished. )r=r>get_query_statusr1is_still_running)rKstatus is_runnings r.is_donezAsyncJob.is_donesP $$**;;DMMJ]]((..??G ~r-c|jjjj|j}|jjjj |S)z Checks the status of the query associated with this instance and returns a bool value indicating whether the query has failed. )r=r>rar1 is_an_errorrKrcs r. is_failedzAsyncJob.is_failedsI $$**;;DMMJ}}""((44V<rar1namerhs r.rczAsyncJob.status s2$$**;;DMMJ{{r-cNtr|jjjddr|ddl}ddl}ddl} |j|j|j|j}d} |j|}|jdds|}|rtd |y|j j#d |jd y#t$rtd|jdwxYw#t|jf$r}dd|d }Yd}~d}~wwxYw) z0Cancels the query associated with this instance.)ENABLE_ASYNC_QUERY_IN_PYTHON_STORED_PROCSFrNzInvalid UUID: ''successzError parsing response: )roerrorz+Failed to cancel query. Returned response: zselect SYSTEM$CANCEL_QUERY('z'))rr=r>"_get_client_side_session_parameter _snowflakejsonuuidUUIDr1 ValueError cancel_queryloadsget TypeErrorJSONDecodeErrorr r@execute)rKrrrsrtraw_cancel_respfailure_responseparsed_cancel_respes r.cancelzAsyncJob.cancels1 # $ ##FF;U    E $--()55dmmDO $  %)ZZ%@")--i?'9$ #ABRAST LL #? b!Q R1 E ?4==/!CDD Et334 $7s;$  s#C7%C=#C:=D$DD$ result_data)snowflake.snowpark.MergeResultsnowflake.snowpark.UpdateResultsnowflake.snowpark.DeleteResultc:|tjk(r>tjj t |ddt |ddS|tj k(r.tjjt |ddSd}d\}}}|jrt |d|}|dz }|jrt |d|}|dz }|jrt |d|}tjj|||S)Nr)rrr) rr) snowflakesnowpark UpdateResultintr* DeleteResultrHrIrJ MergeResult)rKrr4idx rows_inserted rows_updated rows_deleteds r. _table_resultzAsyncJob._table_result?s *11 1%%22KN1%&KN1,=(> ,33 3%%223{1~a7H3IJ JC8? 5M<~~ #KN3$7 8 q}}";q>##67 q}}";q>##67 %%11|\ r-)rrr rrzpandas.DataFramerrrc,|rt|jn |j}|jj |j |j t|j dz D]}|jj|tjtjfvr3|jjd|jjd|tjk(rd} |jjj!|j }|jjj#|sn:t%j&dt(|z|t+t(dz kr|dz }d}n|tjk(r7|jj,j/|jdd d }nY|tjk(r7|jj,j/|jdd d }n|jj1}|jj2|_|tj6k(r#t9||j4|j: }n|tj<k(r#t?||j4|j: }nn|tj@k(r |dd}nR|tjBtjDtjFfvr|jI||}ntK|d |jLD]T}|jj,jN|jPf|jR|jTd |jVV|S)aH Blocks and waits until the query associated with this instance finishes, then returns query results. This acts like executing query in a synchronous way. The data type of returned query results is determined by how you create this :class:`AsyncJob` instance. For example, if this instance is returned by :meth:`DataFrame.collect_nowait`, you will get a list of :class:`Row` s from this method. Args: result_type: Specifies the data type of returned query results. Currently it only supports the following return data types: - "row": returns a list of :class:`Row` objects, which is the same as the return type of :meth:`DataFrame.collect`. - "row_iterator": returns an iterator of :class:`Row` objects, which is the same as the return type of :meth:`DataFrame.to_local_iterator`. - "pandas": returns a ``pandas.DataFrame``, which is the same as the return type of :meth:`DataFrame.to_pandas`. - "pandas_batches": returns an iterator of ``pandas.DataFrame`` s, which is the same as the return type of :meth:`DataFrame.to_pandas_batches`. - "no_result": returns ``None``. You can use this option when you intend to execute the query but don't care about query results (the client will not fetch results either). When you create an :class:`AsyncJob` by :meth:`Session.create_async_job` and retrieve results with this method, ``result_type`` should be specified to determine the result data type. Otherwise, it will return a list of :class:`Row` objects by default. When you create an :class:`AsyncJob` by action methods in :class:`DataFrame` and other classes, ``result_type`` is optional and it will return results with corresponding type. If you still provide a value for it, this value will overwrite the original result data type. Nrz!select * from table(result_scan('z'))rTg?F) to_pandasto_iterdata)r7z is not supported)is_ddl_on_temp_objectr6),rlowerrAr@get_results_from_sfqidr1rErangenextsetr%r&r|sfqidr(r= connectionrarbtimesleepr rVr>_to_data_or_iterfetchall descriptionrGr#rrDr$rr'r)r*r+rrvrB run_queryr^rrCrF) rKr4async_result_type_retry_pattern_posrcr[ractions r.r[zAsyncJob.result]s?`6A [..0 1dFWFW  ++DMM:    +4//!34 ' $$& '! '' --% $$7 8J8J7K3O  0 : : :!" 11BB4==Q}}//@@H -.?@@%,?(@1(DE%*%F "2"9"9 9]]((99 e:F"2"?"? ?]]((99 d:F,,//1K $ 8 8D  $4$8$88+%%#'#7#7 #&6&?&??+%%#'#7#7 #&6&<&<<$Q*" '' '' &&' ++K9JK $5#66G!HII(( F )DMM   ) )  &,&B&B!%!7!7 ""    r-)r9z&snowflake.snowpark.dataframe.DataFrame)r9N)N)r r!r"__doc__rr#rYr rrboolrrMpropertyr2r_rerircrr tupledictrrrrr[r,r-r.r0r0+s|I`)9(<(<.2!&#(,}6  &  tE{+ !  8x}>G =4=  'SR4;T 23&  +  D  z R S z  S  #$ ())    zr-r0)(renumrloggingrtypingrrrrr r snowflake.snowparkrsnowflake.connector.cursorr snowflake.connector.errorsr snowflake.connector.optionsr 4snowflake.snowpark._internal.analyzer.analyzer_utilsr4snowflake.snowpark._internal.analyzer.snowflake_planr"snowflake.snowpark._internal.utilsrrrsnowflake.snowpark.exceptionsrsnowflake.snowpark.functionsrsnowflake.snowpark.rowrsnowflake.snowpark.dataframesnowflake.snowpark.sessionr rWrr0r,r-r.rsh JJ:4.VF ?,&'% H  t llr-