biuRddlmZddlZddlZddlZddlmZ ddl m Z ddlm Z ddlmZddlmZddlmZmZddlmZ dd Z dd Zd Zdd Zdd Z d ddZedddZy)) annotationsN)compute)methods) PANDAS_GE_300)pyarrow_strings_enabled)delayedtokenize) parse_bytesc  ddl} t|ts#tdtt |zdz| t dt|t| j | jjjfs t dtt |z|dkDs| t d| | t d|r |r td | in| } | j|fi| } t|tr| j |n%| j |j|j} | j| d <|dkDr|j|}tj|| fi| }t|dk(rt!j"|d St%rdd lm}||}|j+ddj-|z }||j.dd}|,|| jj1| jj2j5| | jj2j7| j9|j;}tj|| }|j.d\}}|j<d}n$|\}}tj>|j@}|| jj1| jj2jC| j9|j;}tj|| dd}tEtG|ztI|z xsd }|jJdk(rMtMjNtjP||d||z jS|z z}||d<||d<nS|jJdvr+tUjV|||d z|jO}ntdjY|g}|dd|d d}}t[t]||D]z\}\}}|t|d z k(r| |kn| |k}|j_| jja| |k\|}|jctetf|||fd| i| || jit!jj|||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)to_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)6 sqlalchemy isinstancestr TypeErrortype ValueErrorColumnsqlelements ColumnClause create_enginenamelimitpdread_sqllendd from_pandasrdask.dataframe._pyarrowr memory_usagesumilocselectfuncmaxmin select_fromsubquerydtypesSeriesrcountintroundr kindrtolist date_range total_secondsnplinspaceformat enumeratezipwhereand_appendr_read_sql_chunkdispose from_delayed)r&conr divisionsrlimitsbytes_per_chunk head_rowsmetarkwargssaenginerqheadr bytes_per_rowminmaxmaximinirr=partslowersuppersrlowerupperconds P/home/cdr/jupyterlab/.venv/lib/python3.12/site-packages/dask/dataframe/io/sql.pyread_sql_queryresiH c3  -$s)n H I  DEE i#ryy"&&//2N2N!O P @3tICW W   q= <FG G  !4O [STT'/B]M R  c 3] 3F i % ) YYy~~y~~ 6  **F;1} IIi {{1f// t9>>>$A6 6 " $ A%T*D**D*AFFH9T <99Ra=D >  & (>k#,,.) [[F+FQJD$MM'*EJD$IIf%++E   bffkk//67CCCLLNSAKK6*95a8EE%-/+o2NNOPUTU  ::  4$;"="="?+"MNI IaL IbM ZZ? * D$ auMTTVIAAG  Es^Yqr]FF&s66':; >E5!"c&kAo!5u~55= IIbffkk%5.$7 8 $GO $3 ,9 =C    NN ??5$ 22c  ,ddl} ddlm}d| vr+tjdt| j d}d| vr+tjdt| j d}t |ts tdtt|z|C|D]>}t || jtfr td tt|zt |ts#td tt|zd z| in| } | j|fi| }| j}t |tr| j|||| }n tdtt|z|j|rm|Dcgc]a}t |tr)| j||j|jn%| j|j |jcc}n=|jDcgc](}| j|j |j*c}}t |tr)| j||j|jn%| j|j |j}|j |Dcgc]}|j c}vr|j#||j$|j'|}t)d||||||||| | d | Scc}wcc}wcc}w)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_withschema) r&rOr rPrrQrRrSrTr)rr&warningswarnDeprecationWarningpopr r!r"r#r%r)MetaDataTablerMcolumnsr*rKr5r9re) table_namerOr rPrrQrsrRrSrkrTrrUrVr&colrWmcrquerys rdread_sql_tablerys`& e  ZZ(   U  jj j# & 4s4 ;K7L L   CcBIIs#34N$s)n%  c3  -$s)n H I  (/B]M R  c 3] 3F A*c"XXj!6&XQ  4s4 ;K7L L   NN   a% !Z//2778YYqvvqvv./ 2<1C1C DAbii' D  i % )Z// :??@ YYy~~y~~ 6  zz'2Q!&&22u CJJ , ,Z 8E    ' #    - E3sA&L -L +Lc Zt|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!ryre)r&rOr rUs rdr-r-ds7B#sc3 `. 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*rirrk if_existsr index_label chunksizermethodrK)rextras dask_key_namez to_sql-%srN) r r!r$r#dictrr_meta to_delayed _extra_depsr rK dask_compute)rr*rirkrrrrrrrparallelrrU meta_task worker_kwargsrresultlasts rdrrsqz c3 rs"-"08*"  l3f    aH$=N:$      s s s  s  sl $(! !rf