o NrfS@sddlmZddlZddlZddlZddlmZ ddl m Z ddl m Z ddlmZmZddlmZddlmZmZdd lmZ     d$d d Z       d%ddZddZd&ddZd&ddZ          d'd(dd Zedd!d"d#ZdS))) annotationsN)compute)methods) PANDAS_GE_300) from_delayed from_pandas)pyarrow_strings_enabled)delayedtokenize) parse_bytes256 MiBc Ksddl} t|tstdtt|d|durtdt|t| j| jjj fs3tdtt||dksK|dur?td|durK|durKtd|rS|rStd | durYin| } | j |fi| } t|trn| |n| |j |j} | j | d <|dkr| |}t j|| fi| }t|dkrt|d d Strdd lm}m}|||}|jddd|}|dur|jdd}|duri|dur| j| jj| | jj| |}t || }|jd\}}|jd}n |\}}t |j }|dur,| j| jj!| |}t || dd}t"t#||t$|p+d }|j%dkrNt&'t j(||d||)|d}||d<||d<n|j%dvrbt*j+|||d |d'}ntd,|g}|dd|d d}}t-t.||D]5\}\}}|t|d kr| |kn| |k}|/| j0| |k|}|1t2t3|||fd| i| q| 4t5|||dS)ar Read SQL query into a DataFrame. If neither ``divisions`` or ``npartitions`` is given, the memory footprint of the first few rows will be determined, and partitions of size ~256MB will be used. Parameters ---------- sql : SQLAlchemy Selectable SQL query to be executed. TextClause is not supported con : str Full sqlalchemy URI for the database connection index_col : str Column which becomes the index, and defines the partitioning. Should be a indexed column in the SQL server, and any orderable type. If the type is number or time, then partition boundaries can be inferred from ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit ``divisions``. divisions: sequence Values of the index column to split the table by. If given, this will override ``npartitions`` and ``bytes_per_chunk``. The divisions are the value boundaries of the index column used to define the partitions. For example, ``divisions=list('acegikmoqsuwz')`` could be used to partition a string column lexographically into 12 partitions, with the implicit assumption that each partition contains similar numbers of records. npartitions : int Number of partitions, if ``divisions`` is not given. Will split the values of the index column linearly between ``limits``, if given, or the column max/min. The index column must be numeric or time for this to work limits: 2-tuple or None Manually give upper and lower range of values for use with ``npartitions``; if None, first fetches max/min from the DB. Upper limit, if given, is inclusive. bytes_per_chunk : str or int If both ``divisions`` and ``npartitions`` is None, this is the target size of each partition, in bytes head_rows : int How many rows to load for inferring the data-types, and memory per row meta : empty DataFrame or None If provided, do not attempt to infer dtypes, but use these, coercing all chunks on load engine_kwargs : dict or None Specific db engine parameters for sqlalchemy kwargs : dict Additional parameters to pass to `pd.read_sql()` Returns ------- dask.dataframe See Also -------- read_sql_table : Read SQL database table into a DataFrame. rNz'con' must be of type str, not 8Note: Dask does not support SQLAlchemy connectables herez)Must specify index column to partition onz2'index_col' must be of type str or sa.Column, not z'Must provide 'meta' if 'head_rows' is 0z=Must provide 'divisions' or 'npartitions' if 'head_rows' is 0z9Must supply either 'divisions' or 'npartitions', not both index_col) npartitions)check_pyarrow_string_supportedto_pyarrow_stringT)deepindexZmax_1Zcount_1Mz%is)startendfreq)iuf)dtypezwProvided index column is of type "{}". If divisions is not provided the index column type must be numeric or datetime. engine_kwargs) divisions)6 sqlalchemy isinstancestr TypeErrortype ValueErrorColumnsqlelementsZ ColumnClause create_enginenamelimitpdread_sqllenrrZdask.dataframe._pyarrowrrZ memory_usagesumZilocselectfuncmaxmin select_fromZsubquerydtypesZSeriesrcountintroundr kindrtolistZ date_range total_secondsnpZlinspaceformat enumeratezipwhereand_appendr _read_sql_chunkdisposer)r(conrr rlimitsbytes_per_chunk head_rowsmetarkwargssaenginerqheadrrZ bytes_per_rowZminmaxZmaxiZminirr7partsZlowersZuppersrlowerupperZcondrS^/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/dask/dataframe/io/sql.pyread_sql_querysD                 "rUc  sddlddlm} d| vrtdt| dd| vr(tdt| d}tts7tdtt |durR|D]}t|j tfsQtd tt |q=t|tsctd tt |d | duriin| } j |fi| } }ttrj ||| d n tdtt ||rfd d|Dn fddjD}t|tr |j|j n |j|j }|jdd|Dvr||| j|}td||||||||| | d | S)a Read SQL database table into a DataFrame. If neither ``divisions`` or ``npartitions`` is given, the memory footprint of the first few rows will be determined, and partitions of size ~256MB will be used. Parameters ---------- table_name : str Name of SQL table in database. con : str Full sqlalchemy URI for the database connection index_col : str Column which becomes the index, and defines the partitioning. Should be a indexed column in the SQL server, and any orderable type. If the type is number or time, then partition boundaries can be inferred from ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit ``divisions``. columns : sequence of str or SqlAlchemy column or None Which columns to select; if None, gets all. Note can be a mix of str and SqlAlchemy columns schema : str or None Pass this to sqlalchemy to select which DB schema to use within the URI connection divisions: sequence Values of the index column to split the table by. If given, this will override ``npartitions`` and ``bytes_per_chunk``. The divisions are the value boundaries of the index column used to define the partitions. For example, ``divisions=list('acegikmoqsuwz')`` could be used to partition a string column lexographically into 12 partitions, with the implicit assumption that each partition contains similar numbers of records. npartitions : int Number of partitions, if ``divisions`` is not given. Will split the values of the index column linearly between ``limits``, if given, or the column max/min. The index column must be numeric or time for this to work limits: 2-tuple or None Manually give upper and lower range of values for use with ``npartitions``; if None, first fetches max/min from the DB. Upper limit, if given, is inclusive. bytes_per_chunk : str or int If both ``divisions`` and ``npartitions`` is None, this is the target size of each partition, in bytes head_rows : int How many rows to load for inferring the data-types, and memory per row meta : empty DataFrame or None If provided, do not attempt to infer dtypes, but use these, coercing all chunks on load engine_kwargs : dict or None Specific db engine parameters for sqlalchemy kwargs : dict Additional parameters to pass to `pd.read_sql()` Returns ------- dask.dataframe See Also -------- read_sql_query : Read SQL query into a DataFrame. Examples -------- >>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db', ... npartitions=10, index_col='id') # doctest: +SKIP rN)r(tablezWThe `table` keyword has been replaced by `table_name`. Please use `table_name` instead.urizGThe `uri` keyword has been replaced by `con`. Please use `con` instead.z&`table_name` must be of type str, not z8`columns` must be of type List[str], and cannot contain z`con` must be of type str, not r)Z autoload_withschemacs:g|]}t|tr|j|jn|j|jqSrS)r"r#r'columnsr%r+.0crL table_namerSrT Cs z"read_sql_table..csg|] }|j|jqSrS)r'r+r%rZ)rLrSrTr_LscSsg|]}|jqSrS)r+rZrSrSrTr_Ts) r(rFrr rrGrHrIrJrrS)r!r(warningswarnDeprecationWarningpopr"r#r$r%r'r*ZMetaDataTablerErYr+rCr1r5rU)r^rFrr rrGrYrHrIrXrJrrKr(colrMmrqueryrSr]rTread_sql_tablesP            rhcKs2t|trt|||fi|St|||fi|S)a Read SQL query or database table into a DataFrame. This function is a convenience wrapper around ``read_sql_table`` and ``read_sql_query``. It will delegate to the specific function depending on the provided input. A SQL query will be routed to ``read_sql_query``, while a database table name will be routed to ``read_sql_table``. Note that the delegated function might have more specific notes about their functionality not listed here. Parameters ---------- sql : str or SQLAlchemy Selectable Name of SQL table in database or SQL query to be executed. TextClause is not supported con : str Full sqlalchemy URI for the database connection index_col : str Column which becomes the index, and defines the partitioning. Should be a indexed column in the SQL server, and any orderable type. If the type is number or time, then partition boundaries can be inferred from ``npartitions`` or ``bytes_per_chunk``; otherwise must supply explicit ``divisions``. Returns ------- dask.dataframe See Also -------- read_sql_table : Read SQL database table into a DataFrame. read_sql_query : Read SQL query into a DataFrame. )r"r#rhrU)r(rFrrKrSrSrTr.hs !r.cKsddl}|pi}|j|fi|}tj||fi|}|t|dkr'|St|jdkr2|Str6inddi}|j |jfi|S)NrcopyF) r!r*r-r.rEr/r6to_dictrZastype)rNrWrJrrKrLrMdfrSrSrTrDs rDcKsBddl}|pi}|j|fi|}|jdd|i|}||S)NrrFrS)r!r*to_sqlrE)drWrrKrLrMrNrSrSrT _to_sql_chunks rnfailTFr+r#rW if_existsrboolc  st|tstdt|dt||| ||||||| d } tt|jfi| t| dd| r=fdd|D}n)g}}|D] }| t t|fd|id d t |fii|d }qEt|}| rrt |d S|S) aStore Dask Dataframe to a SQL table An empty table is created based on the "meta" DataFrame (and conforming to the caller's "if_exists" preference), and then each block calls pd.DataFrame.to_sql (with `if_exists="append"`). Databases supported by SQLAlchemy [1]_ are supported. Tables can be newly created, appended to, or overwritten. Parameters ---------- name : str Name of SQL table. uri : string Full sqlalchemy URI for the database connection schema : str, optional Specify the schema (if database flavor supports this). If None, use default schema. if_exists : {'fail', 'replace', 'append'}, default 'fail' How to behave if the table already exists. * fail: Raise a ValueError. * replace: Drop the table before inserting new values. * append: Insert new values to the existing table. index : bool, default True Write DataFrame index as a column. Uses `index_label` as the column name in the table. index_label : str or sequence, default None Column label for index column(s). If None is given (default) and `index` is True, then the index names are used. A sequence should be given if the DataFrame uses MultiIndex. chunksize : int, optional Specify the number of rows in each batch to be written at a time. By default, all rows will be written at once. dtype : dict or scalar, optional Specifying the datatype for columns. If a dictionary is used, the keys should be the column names and the values should be the SQLAlchemy types or strings for the sqlite3 legacy mode. If a scalar is provided, it will be applied to all columns. method : {None, 'multi', callable}, optional Controls the SQL insertion clause used: * None : Uses standard SQL ``INSERT`` clause (one per row). * 'multi': Pass multiple values in a single ``INSERT`` clause. * callable with signature ``(pd_table, conn, keys, data_iter)``. Details and a sample callable implementation can be found in the section :ref:`insert method `. compute : bool, default True When true, call dask.compute and perform the load into SQL; otherwise, return a Dask object (or array of per-block objects when parallel=True) parallel : bool, default False When true, have each block append itself to the DB table concurrently. This can result in DB rows being in a different order than the source DataFrame's corresponding rows. When false, load each block into the SQL DB in sequence. engine_kwargs : dict or None Specific db engine parameters for sqlalchemy Raises ------ ValueError When the table already exists and `if_exists` is 'fail' (the default). See Also -------- read_sql : Read a DataFrame from a table. Notes ----- Timezone aware datetime columns will be written as ``Timestamp with timezone`` type with SQLAlchemy if supported by the database. Otherwise, the datetimes will be stored as timezone unaware timestamps local to the original timezone. .. versionadded:: 0.24.0 References ---------- .. [1] https://docs.sqlalchemy.org .. [2] https://www.python.org/dev/peps/pep-0249/ Examples -------- Create a table from scratch with 4 rows. >>> import pandas as pd >>> import dask.dataframe as dd >>> df = pd.DataFrame([ {'i':i, 's':str(i)*2 } for i in range(4) ]) >>> ddf = dd.from_pandas(df, npartitions=2) >>> ddf # doctest: +SKIP Dask DataFrame Structure: i s npartitions=2 0 int64 object 2 ... ... 3 ... ... Dask Name: from_pandas, 2 tasks >>> from dask.utils import tmpfile >>> from sqlalchemy import create_engine, text >>> with tmpfile() as f: ... db = 'sqlite:///%s' %f ... ddf.to_sql('test', db) ... engine = create_engine(db, echo=False) ... with engine.connect() as conn: ... result = conn.execute(text("SELECT * FROM test")).fetchall() >>> result [(0, 0, '00'), (1, 1, '11'), (2, 2, '22'), (3, 3, '33')] z!Expected URI to be a string, got .) r+rWrrXrpr index_label chunksizermethodrC)rpc s:g|]}tt|fdiddt|fiiqS)extras dask_key_name to_sql-%s) _extra_depsrnr )r[rmZ meta_taskZ worker_kwargsrSrTr_Bszto_sql..rvrwrxrN) r"r#r&r%dictr rn_metaZ to_delayedrCryr dask_compute)rkr+rWrXrprrsrtrrurZparallelrrKresultlastrmrSrzrTrlsP }      rl)rvcOs||i|SNrS)r2rvargsrKrSrSrTrycsry)NNNr r NN) NNNNr r NNNr) NroTNNNNTFN)r+r#rWr#rpr#rrq) __future__rr`numpyr=Zpandasr-Z dask.baserr}Zdask.dataframerZdask.dataframe._compatrZdask.dataframe.io.iorrZdask.dataframe.utilsrZ dask.delayedr r Z dask.utilsr rUrhr.rDrnrlryrSrSrSrTs`       8 % '  7