a S/biw @sddlZddlZddlZddlZddlmZddlm Z ddlm Z ddl m Z m Z d#d d Zd$d dZddZd%ddZd&ddZedddZd'ddZd(eeeedddZe dd d!d"ZdS))N)tokenize)delayed)methods) from_delayed from_pandas256 MiBc Kshddl} t|ts*tdtt|d|dur:tdt|t| j| jjj fsftdtt||dks|dur~td|dur|durtd|r|rtd | durin| } | j |fi| } t|tr| |n| |j |j} | j | d <|dkrf| |}t j|| fi| }t|dkr8t|d d S|jd d d|}|durf|jdd}|dur|dur| j| jj| | jj| g|}t || }|jd\}}|jd}n|\}}t |j}|durD| j| jj| g|}t || dd}tt||t j!"|pBd }|j#dkrt$%t j&||d||'|d}||d<||d<n2|j#dvrt()|||d %}ntd*|g}|dd|d d}}t+t,||D]j\}\}}|t|d kr | |kn| |k}|-| j.| |k|}|/t0t1|||fd| i| q| 2t3|||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 here)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_colr npartitionsTZdeepindexmax_1count_1M%iSstartendZfreqiufwProvided index column is of type "{}". If divisions is not provided the index column type must be numeric or datetime. engine_kwargs divisions)4 sqlalchemy isinstancestr TypeErrortype ValueErrorColumnsqlelementsZ ColumnClause create_enginenamelimitpdread_sqllenr memory_usagesumilocselectfuncmaxmin select_fromZsubquerydtypesSeriesdtypecountintrounddaskutils parse_byteskindrtolist date_range total_secondsnplinspaceformat enumeratezipwhereand_appendr_read_sql_chunkdisposer)r*conrr"rlimitsbytes_per_chunk head_rowsmetar kwargssaenginerqhead bytes_per_rowminmaxmaximinir<r=partslowersuppersrloweruppercondre4lib/python3.9/site-packages/dask/dataframe/io/sql.pyread_sql_querysD              "   "rgc  s0ddlddlm} d| vr2tdt| dd| vrPtdt| d}d}ttsxtd t td }|dur|D]$}t|j tfstd td }qt stjd td |st st f||||||||| | | d | St|tst dtt |d| dur"in| } j|fi| }}ttrbj|d || dnt dtt ||rfdd|DnfddjD}t|tr̈ |j|j n |j|j }|jdd|Dvr||| |}tf||||||||| | 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 rNr*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.Fz`read_sql_table` will no longer support {}; please use a `table_name` of type str instead or use `read_sql_query`, if you are using a SQLAlchemy queryTz\`columns` will no longer support SQLAlchemy selectables; please use `read_sql_query` insteadzDask will soon require SQLAlchemy 1.4 or newer. Please update your SQLAlchemy version. Friendly note: Upgrading to SQLAlchemy 1.4 may brake code. Do it with caution. )category) rirjrr"rrRcolumnsrSrTschemarUr z`con` must be of type str, not r ZautoloadZ autoload_withrmz&`table_name` must be of type str, not cs:g|]2}t|tr&|j|jn|j|jqSre)r$r%r)rlr'r-.0crW table_namererf \sz"read_sql_table..csg|]}|j|jqSre)r)r-r'ro)rWrerfrtecSsg|] }|jqSre)r-rorererfrtmru) r*rQrr"rrRrSrTrUr )r#r*warningswarnDeprecationWarningpopr$r%rIr'r)_gt14_old_read_sql_tabler&r,MetaDataTablerPrlr-rNr5r9rg)rsrQrr"rrRrlrSrTrmrUr rVr*Zdeprecated_argscolrXmrqueryrerrrfread_sql_tablesP              rcKs6t|trt|||fi|St|||fi|SdS)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. N)r$r%rrg)r*rQrrVrererfr0s! r0cKs|ddl}|pi}|j|fi|}tj||fi|}|t|dkrN|St|jdkrd|S|j|jddSdS)NrF)copy) r#r,r/r0rPr1r:Zto_dictZastype)rYrjrUr rVrWrXdfrererfrOs rOcKsBddl}|pi}|j|fi|}|jfd|i|}||S)NrrQ)r#r,to_sqlrP)drjr rVrWrXrYrererf _to_sql_chunks r)returncCsPddl}|jdsD|jdsD|jdsD|jdsD|jdrHdSd SdS) ze Check if sqlalchemy.__version__ is at least 1.4.0, when several deprecations were made. rNz0.z1.0z1.1z1.2z1.3FT)r# __version__ startswith)r#rererfrzs     rzc # sddl} ddlm}ddlm}tdt|dur>> df = dd.read_sql_table('accounts', 'sqlite:///path/to/bank.db', ... npartitions=10, index_col='id') # doctest: +SKIP rNrh)r+aYou are using a compatibility version of `read_sql_table` that will be removed in a future version of dask. This function existst to support old versions of SQLAlchemy (< 1.4). This compatibility function is less stable than the new version. We recommend you update your code.r Trnz?Use label when passing an SQLAlchemy instance as the index (%s)z5Must supply either divisions or npartitions, not bothcs$g|]}t|trj|n|qSre)r$r%rlrorirerfrtJruz'_old_read_sql_table..r)rmrrrrz#Must provide meta if head_rows is 0z?Must provide divisions or npartitions when using explicit meta.rrrrrrrrr r!);r#r*Zsqlalchemy.sqlr+rvrwrxr(r,r|r$r%r}rlZLabelr&listrNr-r5r.r9r/r0r1rmrr r2r3r4r6r7r8r:r;r<r=r>r?r@rArBrCrrDrErFrGrHrIrJrKrLrMrrOrPr)#rirjrr"rrRrlrSrTrmrUr rVrWr*r+rXrrrYrZr-r[r\r]r^r<r=r_r`rarrbrcrdrerrfr{sT                     ""r{failTF)r-rj if_existsrc  st|tstdt|dt||| ||||||| d } tt|jfi| t| dd| rzfdd|D}nRg}}|D]@}| t t|fd|id d t |fii|d }qt |}| rt |n|Sd 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 >>> df = pd.DataFrame([ {'i':i, 's':str(i)*2 } for i in range(4) ]) >>> from dask.dataframe import from_pandas >>> ddf = 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 >>> with tmpfile() as f: ... db = 'sqlite:///%s' %f ... ddf.to_sql('test', db) ... engine = create_engine(db, echo=False) ... result = engine.execute("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-rjr rmrr index_label chunksizer<methodrN)rc s:g|]2}tt|fdiddt|fiiqS)extras dask_key_name to_sql-%s) _extra_depsrr)rprZ meta_taskZ worker_kwargsrerfrt<szto_sql..rrrrN)r$r%r(r'dictrrZ_metaZ to_delayedrNrrr@compute)rr-rjrmrrrrr<rrZparallelr rVresultZlastrrerrfrsP|         r)rcOs||i|S)Nre)r6rargsrVrererfr]sr)NNNr r NN) NNNNr r NNN)N)N) NNNNr r NNN) NrTNNNNTFN)rvZnumpyrGZpandasr/r@Z dask.delayedrrriorr rgrr0rOrboolrzr{r%rrrerererfsz    3 F'   U 6