CdtRddlmZddlZddlZddlZddlmZ ddl m Z ddl m Z mZddlmZddlmZmZddlmZ dd Z dd Zd ZddZddZ dd dZedddZdS)!) annotationsN)compute)methods) from_delayed from_pandas)pyarrow_strings_enabled)delayedtokenize) parse_bytes256 MiBc X ddl} t|ts/tdtt |zdz|t dt|t| j| jjj fs,t dtt |z|dks$|t d||t d|r|rtd | in| } | j |fi| } t|tr| |n| |j |j} | j | d <|dkr| |}tj|| fi| }t|dkrt!|d St#rdd lm}m}|||}|dd|z }||jdd}|0|| j| jj| | jj| |}tj|| }|jd\}}|jd}n|\}}tj|j }|| j| jj!| |}tj|| dd}tEtG||ztI|z pd }|j%dkrOtMj'tj(||d||z )|z z}||d<||d<nY|j%dvr.tUj+|||d z|'}n"td,|g}|dd|d d}}t[t]||D]\}\}}|t|d z kr| |kn| |k}|/| j0| |k|}|1tetf|||fd| i| | 4tk|||S)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)deepindexmax_1count_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 ValueErrorColumnsqlelements ColumnClause create_enginenamelimitpdread_sqllenrrdask.dataframe._pyarrowrr memory_usagesumilocselectfuncmaxmin select_fromsubquerydtypesSeriesr!countintroundr kindrtolist date_range total_secondsnplinspaceformat enumeratezipwhereand_appendr _read_sql_chunkdisposer)r+conrr#rlimitsbytes_per_chunk head_rowsmetar"kwargssaenginerqheadrr bytes_per_rowminmaxmaximinir!r@partslowersuppersrlowerupperconds 5lib/python3.11/site-packages/dask/dataframe/io/sql.pyread_sql_queryrfsyH c3    -$s))nn H I   DEEE i#ry"&/2N!O P P  @3tICWCW W    q== <FGG G  !4O U[USTTT'/BB]M R c 3 3] 3 3F i % % 7 ) YYy~y~ 6 6  *F;1}} IIi {1f//// t99>>t333 3 " $ $ +         + * , , ,$$T**D**D*AAFFHH9T <9RaR=D >  && (>(>k#,,..)) [F++FQJD$M'*EEJD$If%%+E   bfk//6677CCCLLNNSSAK6**95a8EE%-/+o2N2NNOOPPUTU  :   4$;"="="?"?+"MNI IaL IbMM Z? * * D$ auMMMTTVVIIAAG  Ess^Yqrr]FF&s66':':;;  >E5!"c&kkAo!5!5u~~55= IIbfkk%5.$77 8 8 $GO $ $3  ,9 =C       NN ty 9 9 99c  rddlddlm} d| vr/tjdt| dd| vr/tjdt| d}t ts,tdttz|M|D]J}t |j tfs,td tt|zKt |ts/td tt|zd z| in| } j |fi| } }t tr ||| n,tdttz||rfd |DnfdjD}t |tr& |j|j n |j|j }|jd|Dvr||| j|}t)d||||||||| | 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) autoload_withschemacg|]]}t|tr&|j|jn|j|j^S)r%r&r*columnsr(r/).0crW table_names re z"read_sql_table..Bsm   a%%/ !Z/27888YYqvqv..    rgcPg|]"}|j|j#Srn)r*r/r()rprqrWs rersz"read_sql_table..Ks+ D D DAbii'' D D Drgcg|] }|j Srn)r/)rprqs rersz"read_sql_table..Ss222Q!&222rg) r+rQrr#rrRrSrTrUr"rn)r$r+warningswarnDeprecationWarningpopr%r&r'r(r*r.MetaDataTablerPror/rNr8r<rf)rrrQrr#rrRrorSrTrlrUr"rVr+colrXmrqueryrWs` @reread_sql_tablersN`& e    ZZ((   U    jj j# & &  4s4 ;K;K7L7L L     CcBIs#344 N$s))nn%  c3    -$s))nn H I   (/BB]M R c 3 3] 3 3F A*c"" XXj!6&XQQ  4s4 ;K;K7L7L L    NN  E          E D D D1C D D D  i % % 7 )Z/ :?@@@ YYy~y~ 6 6  z22'22222u CJ , ,Z 8 8E    ' #      rgc ht|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&rrf)r+rQrrVs rer2r2gsKB#s=c3 <to_dictastype)rYrjrUr"rVrWrXdfs rerOrOs!'RM R c 3 3] 3 3F Q ) )& ) )B NN 2ww!|| T[ " " # #q ( ( yy,,..Uy;;;rgc xddl}|pi}|j|fi|}|jdd|i|}||S)NrrQrn)r$r.to_sqlrP)drjr"rVrWrXrYs re _to_sql_chunkrsc!'RM R c 3 3] 3 3F&&V&v&&A NN HrgfailTFr/r&rj if_existsrboolc Nt|ts tdt|dt ||| |||||||  } t t |jfi| t | d| r"fd|D}n]g}}|D]D}| tt |fd|idd t|fizi|d }Et |}| rt|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 >>> 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"rlrr index_label chunksizer!methodrN)rc ^g|])}tt|fdiddt|fizi*S)extras dask_key_name to_sql-%s) _extra_depsrr )rpr meta_task worker_kwargss rerszto_sql..?s}       !     *HQ,H,H-,H,HH        rgrrrrN) r%r&r)r(dictr r_meta to_delayedrNrr dask_compute)rr/rjrlrrrrr!rrparallelr"rVresultlastrrrs @@rerrsx c3  KIT#YYIIIJJJ  #   F' &&rx::6::I8444M      ]]__      A MM! $  #.0L0Lm0L0L"L    ":DD V__FV rg)rc||i|SNrn)r9rargsrVs rerr`s 4   rg)NNNr r NN) NNNNr r NNNr) NrTNNNNTFN)r/r&rjr&rr&rr) __future__rrvnumpyrGpandasr1 dask.baserrdask.dataframerdask.dataframe.io.iorrdask.dataframe.utilsr dask.delayedr r dask.utilsr rfrr2rOrrrrnrgrers""""""------""""""::::::::888888********""""""  p:p:p:p:n    aaaaH$=$=$=N<<<<"         rrrrrj $(!!!! !!!rg