Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pandas/io/sql.py: 17%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Collection of query wrappers / abstractions to both facilitate data
3retrieval and to reduce dependency on DB-specific API.
4"""
6from __future__ import annotations
8from abc import (
9 ABC,
10 abstractmethod,
11)
12from contextlib import (
13 ExitStack,
14 contextmanager,
15)
16from datetime import (
17 date,
18 datetime,
19 time,
20)
21from functools import partial
22import re
23from typing import (
24 TYPE_CHECKING,
25 Any,
26 Literal,
27 Self,
28 cast,
29 overload,
30)
31import warnings
33import numpy as np
35from pandas._config import using_string_dtype
36from pandas._config.config import _global_config as config
38from pandas._libs import lib
39from pandas.compat._optional import (
40 VERSIONS,
41 import_optional_dependency,
42)
43from pandas.errors import (
44 AbstractMethodError,
45 DatabaseError,
46 Pandas4Warning,
47)
48from pandas.util._decorators import set_module
49from pandas.util._exceptions import find_stack_level
50from pandas.util._validators import check_dtype_backend
52from pandas.core.dtypes.common import (
53 is_dict_like,
54 is_list_like,
55 is_object_dtype,
56 is_string_dtype,
57)
58from pandas.core.dtypes.dtypes import DatetimeTZDtype
59from pandas.core.dtypes.missing import isna
61from pandas.core.api import (
62 DataFrame,
63 Series,
64)
65from pandas.core.arrays import ArrowExtensionArray
66from pandas.core.arrays.string_ import StringDtype
67from pandas.core.base import PandasObject
68import pandas.core.common as com
69from pandas.core.common import maybe_make_list
70from pandas.core.internals.construction import convert_object_array
71from pandas.core.tools.datetimes import (
72 stringify_numeric_column,
73 to_datetime,
74)
76from pandas.io._util import (
77 arrow_table_to_pandas,
78 suppress_pyarrow_values_warning,
79)
81if TYPE_CHECKING:
82 from collections.abc import (
83 Callable,
84 Generator,
85 Iterator,
86 Mapping,
87 )
89 from sqlalchemy import Table
90 from sqlalchemy.sql.expression import (
91 Delete,
92 Select,
93 TextClause,
94 )
96 from pandas._typing import (
97 DtypeArg,
98 DtypeBackend,
99 IndexLabel,
100 )
102 from pandas import Index
104# -----------------------------------------------------------------------------
105# -- Helper functions
108def _process_parse_dates_argument(parse_dates):
109 """Process parse_dates argument for read_sql functions"""
110 # handle non-list entries for parse_dates gracefully
111 if parse_dates is True or parse_dates is None or parse_dates is False:
112 parse_dates = []
114 elif not hasattr(parse_dates, "__iter__"):
115 parse_dates = [parse_dates]
116 return parse_dates
119def _warn_and_stringify_numeric_column(col):
120 # GH#55663 read_sql historically cast integer/float columns to strings so a
121 # ``format`` could be applied to them. That is deprecated in favor of the
122 # user casting explicitly, which lets us drop this shim in a future version.
123 warnings.warn(
124 "Passing a format for integer or float columns to read_sql via "
125 "parse_dates is deprecated. Cast the column to string and apply "
126 "to_datetime after reading, e.g. "
127 "to_datetime(df[col].astype('string'), format=...).",
128 Pandas4Warning,
129 stacklevel=find_stack_level(),
130 )
131 return stringify_numeric_column(col)
134def _handle_date_column(
135 col, utc: bool = False, format: str | dict[str, Any] | None = None
136):
137 if isinstance(format, dict):
138 # GH35185 Allow custom error values in parse_dates argument of
139 # read_sql like functions.
140 # Format can take on custom to_datetime argument values such as
141 # {"errors": "coerce"} or {"dayfirst": True}
142 if format.get("format") is not None and col.dtype.kind in "iuf":
143 col = _warn_and_stringify_numeric_column(col)
144 return to_datetime(col, **format)
145 else:
146 # Allow passing of formatting string for integers
147 # GH17855
148 if format is None and (
149 issubclass(col.dtype.type, np.floating)
150 or issubclass(col.dtype.type, np.integer)
151 ):
152 format = "s"
153 if format in ["D", "d", "h", "m", "s", "ms", "us", "ns"]:
154 return to_datetime(col, errors="coerce", unit=format, utc=utc)
155 elif isinstance(col.dtype, DatetimeTZDtype):
156 # coerce to UTC timezone
157 # GH11216
158 return to_datetime(col, utc=True)
159 else:
160 if format is not None and col.dtype.kind in "iuf":
161 col = _warn_and_stringify_numeric_column(col)
162 return to_datetime(col, errors="coerce", format=format, utc=utc)
165def _parse_date_columns(data_frame: DataFrame, parse_dates) -> DataFrame:
166 """
167 Force non-datetime columns to be read as such.
168 Supports both string formatted and integer timestamp columns.
169 """
170 parse_dates = _process_parse_dates_argument(parse_dates)
172 # we want to coerce datetime64_tz dtypes for now to UTC
173 # we could in theory do a 'nice' conversion from a FixedOffset tz
174 # GH11216
175 for i, (col_name, df_col) in enumerate(data_frame.items()):
176 if isinstance(df_col.dtype, DatetimeTZDtype) or col_name in parse_dates:
177 try:
178 fmt = parse_dates[col_name]
179 except (KeyError, TypeError):
180 fmt = None
181 data_frame.isetitem(i, _handle_date_column(df_col, format=fmt))
183 return data_frame
186def _convert_arrays_to_dataframe(
187 data,
188 columns,
189 coerce_float: bool = True,
190 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
191) -> DataFrame:
192 if data and is_dict_like(data[0]) and not isinstance(data[0], (list, tuple)):
193 # DBAPI cursors returning dict rows, e.g. pymysql's DictCursor. Use
194 # values() rather than lookup by column name, since such cursors may
195 # rename duplicate columns (GH#53028). Skip list-based rows like
196 # psycopg2's DictRow, whose values() drops duplicate columns.
197 data = [tuple(row.values()) for row in data]
198 content = lib.to_object_array_tuples(data)
199 idx_len = content.shape[0]
200 arrays = convert_object_array(
201 list(content.T),
202 dtype=None,
203 coerce_float=coerce_float,
204 dtype_backend=dtype_backend,
205 )
206 if dtype_backend == "pyarrow":
207 pa = import_optional_dependency("pyarrow")
209 result_arrays = []
210 for arr in arrays:
211 pa_array = pa.array(arr, from_pandas=True)
212 if arr.dtype == "string":
213 # TODO: Arrow still infers strings arrays as regular strings instead
214 # of large_string, which is what we preserver everywhere else for
215 # dtype_backend="pyarrow". We may want to reconsider this
216 pa_array = pa_array.cast(pa.string())
217 result_arrays.append(ArrowExtensionArray(pa_array))
218 arrays = result_arrays # type: ignore[assignment]
219 if arrays:
220 return DataFrame._from_arrays(
221 arrays, columns=columns, index=range(idx_len), verify_integrity=False
222 )
223 else:
224 return DataFrame(columns=columns)
227def _wrap_result(
228 data,
229 columns,
230 index_col=None,
231 coerce_float: bool = True,
232 parse_dates=None,
233 dtype: DtypeArg | None = None,
234 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
235) -> DataFrame:
236 """Wrap result set of a SQLAlchemy query in a DataFrame."""
237 frame = _convert_arrays_to_dataframe(data, columns, coerce_float, dtype_backend)
239 if dtype:
240 frame = frame.astype(dtype)
242 frame = _parse_date_columns(frame, parse_dates)
244 if index_col is not None:
245 frame = frame.set_index(index_col)
247 return frame
250def _wrap_result_adbc(
251 df: DataFrame,
252 *,
253 index_col=None,
254 parse_dates=None,
255 dtype: DtypeArg | None = None,
256 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
257) -> DataFrame:
258 """Wrap result set of a SQLAlchemy query in a DataFrame."""
259 if dtype:
260 df = df.astype(dtype)
262 df = _parse_date_columns(df, parse_dates)
264 if index_col is not None:
265 df = df.set_index(index_col)
267 return df
270# -----------------------------------------------------------------------------
271# -- Read and write to DataFrames
274@overload
275def read_sql_table( # pyright: ignore[reportOverlappingOverload]
276 table_name: str,
277 con,
278 schema=...,
279 index_col: str | list[str] | None = ...,
280 coerce_float=...,
281 parse_dates: list[str] | dict[str, str] | dict[str, dict[str, Any]] | None = ...,
282 columns: list[str] | None = ...,
283 chunksize: None = ...,
284 dtype_backend: DtypeBackend | lib.NoDefault = ...,
285) -> DataFrame: ...
288@overload
289def read_sql_table(
290 table_name: str,
291 con,
292 schema=...,
293 index_col: str | list[str] | None = ...,
294 coerce_float=...,
295 parse_dates: list[str] | dict[str, str] | dict[str, dict[str, Any]] | None = ...,
296 columns: list[str] | None = ...,
297 chunksize: int = ...,
298 dtype_backend: DtypeBackend | lib.NoDefault = ...,
299) -> Iterator[DataFrame]: ...
302@set_module("pandas")
303def read_sql_table(
304 table_name: str,
305 con,
306 schema: str | None = None,
307 index_col: str | list[str] | None = None,
308 coerce_float: bool = True,
309 parse_dates: list[str] | dict[str, str] | dict[str, dict[str, Any]] | None = None,
310 columns: list[str] | None = None,
311 chunksize: int | None = None,
312 dtype_backend: DtypeBackend | lib.NoDefault = lib.no_default,
313) -> DataFrame | Iterator[DataFrame]:
314 """
315 Read SQL database table into a DataFrame.
317 Given a table name and a SQLAlchemy connectable, returns a DataFrame.
318 This function does not support DBAPI connections.
320 Parameters
321 ----------
322 table_name : str
323 Name of SQL table in database.
324 con : SQLAlchemy connectable or str
325 A database URI could be provided as str.
326 SQLite DBAPI connection mode not supported.
327 schema : str, default None
328 Name of SQL schema in database to query (if database flavor
329 supports this). Uses default schema if None (default).
330 index_col : str or list of str, optional, default: None
331 Column(s) to set as index(MultiIndex).
332 coerce_float : bool, default True
333 Attempts to convert values of non-string, non-numeric objects (like
334 decimal.Decimal) to floating point. Can result in loss of Precision.
335 parse_dates : list or dict, default None
336 - List of column names to parse as dates.
337 - Dict of ``{column_name: format string}`` where format string is
338 strftime compatible in case of parsing string times or is one of
339 (D, s, ns, ms, us) in case of parsing integer timestamps.
340 - Dict of ``{column_name: arg dict}``, where the arg dict corresponds
341 to the keyword arguments of :func:`pandas.to_datetime`
342 Especially useful with databases without native Datetime support,
343 such as SQLite.
344 columns : list, default None
345 List of column names to select from SQL table.
346 chunksize : int, default None
347 If specified, returns an iterator where `chunksize` is the number of
348 rows to include in each chunk. By itself this typically does not
349 reduce peak memory usage, as most drivers buffer the full result set
350 unless a server-side cursor is used; see the :ref:`user guide
351 <io.sql.chunksize>` on streaming results.
352 dtype_backend : {'numpy_nullable', 'pyarrow'}
353 Back-end data type applied to the resultant :class:`DataFrame`
354 (still experimental). If not specified, the default behavior
355 is to not use nullable data types. If specified, the behavior
356 is as follows:
358 * ``"numpy_nullable"``: returns nullable-dtype-backed :class:`DataFrame`
359 * ``"pyarrow"``: returns pyarrow-backed nullable
360 :class:`ArrowDtype` :class:`DataFrame`
362 .. versionadded:: 2.0
364 Returns
365 -------
366 DataFrame or Iterator[DataFrame]
367 A SQL table is returned as two-dimensional data structure with labeled
368 axes.
370 See Also
371 --------
372 read_sql_query : Read SQL query into a DataFrame.
373 read_sql : Read SQL query or database table into a DataFrame.
375 Notes
376 -----
377 Any datetime values with time zone information will be converted to UTC.
379 Examples
380 --------
381 >>> pd.read_sql_table("table_name", "postgres:///db_name") # doctest:+SKIP
382 """
384 check_dtype_backend(dtype_backend)
385 if dtype_backend is lib.no_default:
386 dtype_backend = "numpy" # type: ignore[assignment]
387 assert dtype_backend is not lib.no_default
389 with pandasSQL_builder(con, schema=schema, need_transaction=True) as pandas_sql:
390 if not pandas_sql.has_table(table_name):
391 raise ValueError(f"Table {table_name} not found")
393 table = pandas_sql.read_table(
394 table_name,
395 index_col=index_col,
396 coerce_float=coerce_float,
397 parse_dates=parse_dates,
398 columns=columns,
399 chunksize=chunksize,
400 dtype_backend=dtype_backend,
401 )
403 if table is not None:
404 return table
405 else:
406 raise ValueError(f"Table {table_name} not found", con)
409@overload
410def read_sql_query( # pyright: ignore[reportOverlappingOverload]
411 sql,
412 con,
413 index_col: str | list[str] | None = ...,
414 coerce_float=...,
415 params: list[Any] | Mapping[str, Any] | None = ...,
416 parse_dates: list[str] | dict[str, str] | dict[str, dict[str, Any]] | None = ...,
417 chunksize: None = ...,
418 dtype: DtypeArg | None = ...,
419 dtype_backend: DtypeBackend | lib.NoDefault = ...,
420) -> DataFrame: ...
423@overload
424def read_sql_query(
425 sql,
426 con,
427 index_col: str | list[str] | None = ...,
428 coerce_float=...,
429 params: list[Any] | Mapping[str, Any] | None = ...,
430 parse_dates: list[str] | dict[str, str] | dict[str, dict[str, Any]] | None = ...,
431 chunksize: int = ...,
432 dtype: DtypeArg | None = ...,
433 dtype_backend: DtypeBackend | lib.NoDefault = ...,
434) -> Iterator[DataFrame]: ...
437@set_module("pandas")
438def read_sql_query(
439 sql,
440 con,
441 index_col: str | list[str] | None = None,
442 coerce_float: bool = True,
443 params: list[Any] | Mapping[str, Any] | None = None,
444 parse_dates: list[str] | dict[str, str] | dict[str, dict[str, Any]] | None = None,
445 chunksize: int | None = None,
446 dtype: DtypeArg | None = None,
447 dtype_backend: DtypeBackend | lib.NoDefault = lib.no_default,
448) -> DataFrame | Iterator[DataFrame]:
449 """
450 Read SQL query into a DataFrame.
452 Returns a DataFrame corresponding to the result set of the query
453 string. Optionally provide an `index_col` parameter to use one of the
454 columns as the index, otherwise default integer index will be used.
456 Parameters
457 ----------
458 sql : str SQL query or SQLAlchemy Selectable (select or text object)
459 SQL query to be executed.
460 con : SQLAlchemy connectable, str, or sqlite3 connection
461 Using SQLAlchemy makes it possible to use any DB supported by that
462 library. If a DBAPI2 object, only sqlite3 is supported.
463 index_col : str or list of str, optional, default: None
464 Column(s) to set as index(MultiIndex).
465 coerce_float : bool, default True
466 Attempts to convert values of non-string, non-numeric objects (like
467 decimal.Decimal) to floating point. Useful for SQL result sets. This
468 can lose precision: an integral ``decimal.Decimal`` larger than ``2**53``
469 has no exact ``float64`` representation, so a long identifier can be
470 silently rounded. Pass ``False`` to leave such values as Python objects
471 in an ``object``-dtype column.
472 params : list, tuple or mapping, optional, default: None
473 List of parameters to pass to execute method. The syntax used
474 to pass parameters is database driver dependent. Check your
475 database driver documentation for which of the five syntax styles,
476 described in PEP 249's paramstyle, is supported.
477 Eg. for psycopg2, uses %(name)s so use params={'name' : 'value'}.
478 parse_dates : list or dict, default: None
479 - List of column names to parse as dates.
480 - Dict of ``{column_name: format string}`` where format string is
481 strftime compatible in case of parsing string times, or is one of
482 (D, s, ns, ms, us) in case of parsing integer timestamps.
483 - Dict of ``{column_name: arg dict}``, where the arg dict corresponds
484 to the keyword arguments of :func:`pandas.to_datetime`
485 Especially useful with databases without native Datetime support,
486 such as SQLite.
487 chunksize : int, default None
488 If specified, return an iterator where `chunksize` is the number of
489 rows to include in each chunk. By itself this typically does not
490 reduce peak memory usage, as most drivers buffer the full result set
491 unless a server-side cursor is used; see the :ref:`user guide
492 <io.sql.chunksize>` on streaming results.
493 dtype : Type name or dict of columns
494 Data type for data or columns. E.g. np.float64 or
495 {'a': np.float64, 'b': np.int32, 'c': 'Int64'}.
496 dtype_backend : {'numpy_nullable', 'pyarrow'}
497 Back-end data type applied to the resultant :class:`DataFrame`
498 (still experimental). If not specified, the default behavior
499 is to not use nullable data types. If specified, the behavior
500 is as follows:
502 * ``"numpy_nullable"``: returns nullable-dtype-backed :class:`DataFrame`
503 * ``"pyarrow"``: returns pyarrow-backed nullable
504 :class:`ArrowDtype` :class:`DataFrame`
506 .. versionadded:: 2.0
508 Returns
509 -------
510 DataFrame or Iterator[DataFrame]
511 Returns a DataFrame object that contains the result set of the
512 executed SQL query, in relation to the specified database connection.
514 See Also
515 --------
516 read_sql_table : Read SQL database table into a DataFrame.
517 read_sql : Read SQL query or database table into a DataFrame.
519 Notes
520 -----
521 Any datetime values with time zone information parsed via the `parse_dates`
522 parameter will be converted to UTC.
524 Examples
525 --------
526 >>> from sqlalchemy import create_engine # doctest: +SKIP
527 >>> engine = create_engine("sqlite:///database.db") # doctest: +SKIP
528 >>> sql_query = "SELECT int_column FROM test_data" # doctest: +SKIP
529 >>> with engine.connect() as conn, conn.begin(): # doctest: +SKIP
530 ... data = pd.read_sql_query(sql_query, conn) # doctest: +SKIP
531 """
533 check_dtype_backend(dtype_backend)
534 if dtype_backend is lib.no_default:
535 dtype_backend = "numpy" # type: ignore[assignment]
536 assert dtype_backend is not lib.no_default
538 with pandasSQL_builder(con) as pandas_sql:
539 return pandas_sql.read_query(
540 sql,
541 index_col=index_col,
542 params=params,
543 coerce_float=coerce_float,
544 parse_dates=parse_dates,
545 chunksize=chunksize,
546 dtype=dtype,
547 dtype_backend=dtype_backend,
548 )
551@overload
552def read_sql( # pyright: ignore[reportOverlappingOverload]
553 sql,
554 con,
555 index_col: str | list[str] | None = ...,
556 coerce_float=...,
557 params=...,
558 parse_dates=...,
559 columns: list[str] = ...,
560 chunksize: None = ...,
561 dtype_backend: DtypeBackend | lib.NoDefault = ...,
562 dtype: DtypeArg | None = None,
563) -> DataFrame: ...
566@overload
567def read_sql(
568 sql,
569 con,
570 index_col: str | list[str] | None = ...,
571 coerce_float=...,
572 params=...,
573 parse_dates=...,
574 columns: list[str] = ...,
575 chunksize: int = ...,
576 dtype_backend: DtypeBackend | lib.NoDefault = ...,
577 dtype: DtypeArg | None = None,
578) -> Iterator[DataFrame]: ...
581@set_module("pandas")
582def read_sql(
583 sql,
584 con,
585 index_col: str | list[str] | None = None,
586 coerce_float: bool = True,
587 params=None,
588 parse_dates=None,
589 columns: list[str] | None = None,
590 chunksize: int | None = None,
591 dtype_backend: DtypeBackend | lib.NoDefault = lib.no_default,
592 dtype: DtypeArg | None = None,
593) -> DataFrame | Iterator[DataFrame]:
594 """
595 Read SQL query or database table into a DataFrame.
597 This function is a convenience wrapper around ``read_sql_table`` and
598 ``read_sql_query`` (for backward compatibility). It will delegate
599 to the specific function depending on the provided input. A SQL query
600 will be routed to ``read_sql_query``, while a database table name will
601 be routed to ``read_sql_table``. Note that the delegated function might
602 have more specific notes about their functionality not listed here.
604 Parameters
605 ----------
606 sql : str or SQLAlchemy Selectable (select or text object)
607 SQL query to be executed or a table name.
608 con : ADBC Connection, SQLAlchemy connectable, str, or sqlite3 connection
609 ADBC provides high performance I/O with native type support, where available.
610 Using SQLAlchemy makes it possible to use any DB supported by that
611 library. If a DBAPI2 object, only sqlite3 is supported. The user is responsible
612 for engine disposal and connection closure for the ADBC connection and
613 SQLAlchemy connectable; str connections are closed automatically. See
614 `here <https://docs.sqlalchemy.org/en/20/core/connections.html>`_.
616 .. versionadded:: 2.2.0
617 Support for ADBC drivers.
618 index_col : str or list of str, optional, default: None
619 Column(s) to set as index(MultiIndex).
620 coerce_float : bool, default True
621 Attempts to convert values of non-string, non-numeric objects (like
622 decimal.Decimal) to floating point, useful for SQL result sets. This
623 can lose precision: an integral ``decimal.Decimal`` larger than ``2**53``
624 has no exact ``float64`` representation, so a long identifier can be
625 silently rounded. Pass ``False`` to leave such values as Python objects
626 in an ``object``-dtype column.
627 params : list, tuple or dict, optional, default: None
628 List of parameters to pass to execute method. The syntax used
629 to pass parameters is database driver dependent. Check your
630 database driver documentation for which of the five syntax styles,
631 described in PEP 249's paramstyle, is supported.
632 Eg. for psycopg2, uses %(name)s so use params={'name' : 'value'}.
633 parse_dates : list or dict, default: None
634 - List of column names to parse as dates.
635 - Dict of ``{column_name: format string}`` where format string is
636 strftime compatible in case of parsing string times, or is one of
637 (D, s, ns, ms, us) in case of parsing integer timestamps.
638 - Dict of ``{column_name: arg dict}``, where the arg dict corresponds
639 to the keyword arguments of :func:`pandas.to_datetime`
640 Especially useful with databases without native Datetime support,
641 such as SQLite.
642 columns : list, default: None
643 List of column names to select from SQL table (only used when reading
644 a table).
645 chunksize : int, default None
646 If specified, return an iterator where `chunksize` is the
647 number of rows to include in each chunk. By itself this typically
648 does not reduce peak memory usage, as most drivers buffer the full
649 result set unless a server-side cursor is used; see the
650 :ref:`user guide <io.sql.chunksize>` on streaming results.
651 dtype_backend : {'numpy_nullable', 'pyarrow'}
652 Back-end data type applied to the resultant :class:`DataFrame`
653 (still experimental). If not specified, the default behavior
654 is to not use nullable data types. If specified, the behavior
655 is as follows:
657 * ``"numpy_nullable"``: returns nullable-dtype-backed :class:`DataFrame`
658 * ``"pyarrow"``: returns pyarrow-backed nullable
659 :class:`ArrowDtype` :class:`DataFrame`
661 .. versionadded:: 2.0
662 dtype : Type name or dict of columns
663 Data type for data or columns. E.g. np.float64 or
664 {'a': np.float64, 'b': np.int32, 'c': 'Int64'}.
665 The argument is ignored if a table is passed instead of a query.
667 .. versionadded:: 2.0.0
669 Returns
670 -------
671 DataFrame or Iterator[DataFrame]
672 Returns a DataFrame object that contains the result set of the
673 executed SQL query or an SQL Table based on the provided input,
674 in relation to the specified database connection.
676 See Also
677 --------
678 read_sql_table : Read SQL database table into a DataFrame.
679 read_sql_query : Read SQL query into a DataFrame.
681 Notes
682 -----
683 ``pandas`` does not attempt to sanitize SQL statements;
684 instead it simply forwards the statement you are executing
685 to the underlying driver, which may or may not sanitize from there.
686 Please refer to the underlying driver documentation for any details.
687 Generally, be wary when accepting statements from arbitrary sources.
689 Examples
690 --------
691 Read data from SQL via either a SQL query or a SQL tablename.
692 When using a SQLite database only SQL queries are accepted,
693 providing only the SQL tablename will result in an error.
695 >>> from sqlite3 import connect
696 >>> conn = connect(":memory:")
697 >>> df = pd.DataFrame(
698 ... data=[[0, "10/11/12"], [1, "12/11/10"]],
699 ... columns=["int_column", "date_column"],
700 ... )
701 >>> df.to_sql(name="test_data", con=conn)
702 2
704 >>> pd.read_sql("SELECT int_column, date_column FROM test_data", conn)
705 int_column date_column
706 0 0 10/11/12
707 1 1 12/11/10
709 >>> pd.read_sql("test_data", "postgres:///db_name") # doctest:+SKIP
711 For parameterized query, using ``params`` is recommended over string interpolation.
713 >>> from sqlalchemy import text
714 >>> sql = text(
715 ... "SELECT int_column, date_column FROM test_data WHERE int_column=:int_val"
716 ... )
717 >>> pd.read_sql(sql, conn, params={"int_val": 1}) # doctest:+SKIP
718 int_column date_column
719 0 1 12/11/10
721 Apply date parsing to columns through the ``parse_dates`` argument
722 The ``parse_dates`` argument calls ``pd.to_datetime`` on the provided columns.
723 Custom argument values for applying ``pd.to_datetime`` on a column are specified
724 via a dictionary format:
726 >>> pd.read_sql(
727 ... "SELECT int_column, date_column FROM test_data",
728 ... conn,
729 ... parse_dates={"date_column": {"format": "%d/%m/%y"}},
730 ... )
731 int_column date_column
732 0 0 2012-11-10
733 1 1 2010-11-12
735 pandas supports reading via ADBC drivers:
737 >>> from adbc_driver_postgresql import dbapi # doctest:+SKIP
738 >>> with dbapi.connect("postgres:///db_name") as conn: # doctest:+SKIP
739 ... pd.read_sql("SELECT int_column FROM test_data", conn)
740 int_column
741 0 0
742 1 1
743 """
745 check_dtype_backend(dtype_backend)
746 if dtype_backend is lib.no_default:
747 dtype_backend = "numpy" # type: ignore[assignment]
748 assert dtype_backend is not lib.no_default
750 with pandasSQL_builder(con) as pandas_sql:
751 if isinstance(pandas_sql, SQLiteDatabase):
752 return pandas_sql.read_query(
753 sql,
754 index_col=index_col,
755 params=params,
756 coerce_float=coerce_float,
757 parse_dates=parse_dates,
758 chunksize=chunksize,
759 dtype_backend=dtype_backend,
760 dtype=dtype,
761 )
763 try:
764 _is_table_name = pandas_sql.has_table(sql)
765 except Exception:
766 # using generic exception to catch errors from sql drivers (GH24988)
767 _is_table_name = False
769 if _is_table_name:
770 return pandas_sql.read_table(
771 sql,
772 index_col=index_col,
773 coerce_float=coerce_float,
774 parse_dates=parse_dates,
775 columns=columns,
776 chunksize=chunksize,
777 dtype_backend=dtype_backend,
778 )
779 else:
780 return pandas_sql.read_query(
781 sql,
782 index_col=index_col,
783 params=params,
784 coerce_float=coerce_float,
785 parse_dates=parse_dates,
786 chunksize=chunksize,
787 dtype_backend=dtype_backend,
788 dtype=dtype,
789 )
792def to_sql(
793 frame,
794 name: str,
795 con,
796 schema: str | None = None,
797 if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
798 index: bool = True,
799 index_label: IndexLabel | None = None,
800 chunksize: int | None = None,
801 dtype: DtypeArg | None = None,
802 method: Literal["multi"] | Callable | None = None,
803 engine: str = "auto",
804 **engine_kwargs,
805) -> int | None:
806 """
807 Write records stored in a DataFrame to a SQL database.
809 .. warning::
810 The pandas library does not attempt to sanitize inputs provided via a to_sql call.
811 Please refer to the documentation for the underlying database driver to see if it
812 will properly prevent injection, or alternatively be advised of a security risk when
813 executing arbitrary commands in a to_sql call.
815 Parameters
816 ----------
817 frame : DataFrame, Series
818 name : str
819 Name of SQL table.
820 con : ADBC Connection, SQLAlchemy connectable, str, or sqlite3 connection
821 or sqlite3 DBAPI2 connection
822 ADBC provides high performance I/O with native type support, where available.
823 Using SQLAlchemy makes it possible to use any DB supported by that
824 library.
825 If a DBAPI2 object, only sqlite3 is supported.
826 schema : str, optional
827 Name of SQL schema in database to write to (if database flavor
828 supports this). If None, use default schema (default).
829 if_exists : {'fail', 'replace', 'append', 'delete_rows'}, default 'fail'
830 - fail: If table exists, do nothing.
831 - replace: If table exists, drop it, recreate it, and insert data.
832 - append: If table exists, insert data. Create if does not exist.
833 - delete_rows: If a table exists, delete all records and insert data.
834 index : bool, default True
835 Write DataFrame index as a column.
836 index_label : str or sequence, optional
837 Column label for index column(s). If None is given (default) and
838 `index` is True, then the index names are used.
839 A sequence should be given if the DataFrame uses MultiIndex.
840 chunksize : int, optional
841 Specify the number of rows in each batch to be written at a time.
842 By default, all rows will be written at once.
843 dtype : dict or scalar, optional
844 Specifying the datatype for columns. If a dictionary is used, the
845 keys should be the column names and the values should be the
846 SQLAlchemy types or strings for the sqlite3 fallback mode. If a
847 scalar is provided, it will be applied to all columns.
848 method : {None, 'multi', callable}, optional
849 Controls the SQL insertion clause used:
851 - None : Uses standard SQL ``INSERT`` clause (one per row).
852 - ``'multi'``: Pass multiple values in a single ``INSERT`` clause.
853 - callable with signature ``(pd_table, conn, keys, data_iter) -> int | None``.
855 Details and a sample callable implementation can be found in the
856 section :ref:`insert method <io.sql.method>`.
857 engine : {'auto', 'sqlalchemy'}, default 'auto'
858 SQL engine library to use. If 'auto', then the option
859 ``io.sql.engine`` is used. The default ``io.sql.engine``
860 behavior is 'sqlalchemy'
862 **engine_kwargs
863 Any additional kwargs are passed to the engine.
865 Returns
866 -------
867 None or int
868 Number of rows affected by to_sql. None is returned if the callable
869 passed into ``method`` does not return an integer number of rows.
871 Notes
872 -----
873 The returned rows affected is the sum of the ``rowcount`` attribute of ``sqlite3.Cursor``
874 or SQLAlchemy connectable. If using ADBC the returned rows are the result
875 of ``Cursor.adbc_ingest``. The returned value may not reflect the exact number of written
876 rows as stipulated in the
877 `sqlite3 <https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.rowcount>`__ or
878 `SQLAlchemy <https://docs.sqlalchemy.org/en/14/core/connections.html#sqlalchemy.engine.BaseCursorResult.rowcount>`__
879 """ # noqa: E501
880 if if_exists not in ("fail", "replace", "append", "delete_rows"):
881 raise ValueError(f"'{if_exists}' is not valid for if_exists")
883 if isinstance(frame, Series):
884 frame = frame.to_frame()
885 elif not isinstance(frame, DataFrame):
886 raise NotImplementedError(
887 "'frame' argument should be either a Series or a DataFrame"
888 )
890 with pandasSQL_builder(con, schema=schema, need_transaction=True) as pandas_sql:
891 return pandas_sql.to_sql(
892 frame,
893 name,
894 if_exists=if_exists,
895 index=index,
896 index_label=index_label,
897 schema=schema,
898 chunksize=chunksize,
899 dtype=dtype,
900 method=method,
901 engine=engine,
902 **engine_kwargs,
903 )
906def has_table(table_name: str, con, schema: str | None = None) -> bool:
907 """
908 Check if DataBase has named table.
910 Parameters
911 ----------
912 table_name: string
913 Name of SQL table.
914 con: ADBC Connection, SQLAlchemy connectable, str, or sqlite3 connection
915 ADBC provides high performance I/O with native type support, where available.
916 Using SQLAlchemy makes it possible to use any DB supported by that
917 library.
918 If a DBAPI2 object, only sqlite3 is supported.
919 schema : string, default None
920 Name of SQL schema in database to write to (if database flavor supports
921 this). If None, use default schema (default).
923 Returns
924 -------
925 boolean
926 """
927 with pandasSQL_builder(con, schema=schema) as pandas_sql:
928 return pandas_sql.has_table(table_name)
931table_exists = has_table
934def pandasSQL_builder(
935 con,
936 schema: str | None = None,
937 need_transaction: bool = False,
938) -> PandasSQL:
939 """
940 Convenience function to return the correct PandasSQL subclass based on the
941 provided parameters. Also creates a sqlalchemy connection and transaction
942 if necessary.
943 """
944 import sqlite3
946 if isinstance(con, sqlite3.Connection) or con is None:
947 return SQLiteDatabase(con)
949 sqlalchemy = import_optional_dependency("sqlalchemy", errors="ignore")
951 if isinstance(con, str) and sqlalchemy is None:
952 raise ImportError(
953 "Using a URI string requires 'sqlalchemy' version "
954 f"'{VERSIONS['sqlalchemy']}' or newer."
955 )
957 if sqlalchemy is not None and isinstance(con, (str, sqlalchemy.engine.Connectable)):
958 return SQLDatabase(con, schema, need_transaction)
960 adbc = import_optional_dependency("adbc_driver_manager.dbapi", errors="ignore")
961 if adbc and isinstance(con, adbc.Connection):
962 return ADBCDatabase(con)
964 warnings.warn(
965 "pandas only supports SQLAlchemy connectable (engine/connection) or "
966 "database string URI or sqlite3 DBAPI2 connection. Other DBAPI2 "
967 "objects are not tested. Please consider using SQLAlchemy.",
968 UserWarning,
969 stacklevel=find_stack_level(),
970 )
971 return SQLiteDatabase(con)
974class SQLTable(PandasObject):
975 """
976 For mapping Pandas tables to SQL tables.
977 Uses fact that table is reflected by SQLAlchemy to
978 do better type conversions.
979 Also holds various flags needed to avoid having to
980 pass them between functions all the time.
981 """
983 # TODO: support for multiIndex
985 def __init__(
986 self,
987 name: str,
988 pandas_sql_engine,
989 frame=None,
990 index: bool | str | list[str] | None = True,
991 if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
992 prefix: str = "pandas",
993 index_label=None,
994 schema=None,
995 keys=None,
996 dtype: DtypeArg | None = None,
997 ) -> None:
998 self.name = name
999 self.pd_sql = pandas_sql_engine
1000 self.prefix = prefix
1001 self.frame = frame
1002 self.index = self._index_name(index, index_label)
1003 self.schema = schema
1004 self.if_exists = if_exists
1005 self.keys = keys
1006 self.dtype = dtype
1008 if frame is not None:
1009 # We want to initialize based on a dataframe
1010 self.table = self._create_table_setup()
1011 else:
1012 # no data provided, read-only mode
1013 self.table = self.pd_sql.get_table(self.name, self.schema)
1015 if self.table is None:
1016 raise ValueError(f"Could not init table '{name}'")
1018 if not len(self.name):
1019 raise ValueError("Empty table name specified")
1021 def exists(self):
1022 return self.pd_sql.has_table(self.name, self.schema)
1024 def sql_schema(self) -> str:
1025 from sqlalchemy.schema import CreateTable
1027 return str(CreateTable(self.table).compile(self.pd_sql.con))
1029 def _execute_create(self) -> None:
1030 # Inserting table into database, add to MetaData object
1031 self.table = self.table.to_metadata(self.pd_sql.meta)
1032 with self.pd_sql.run_transaction():
1033 self.table.create(bind=self.pd_sql.con)
1035 def create(self) -> None:
1036 if self.exists():
1037 if self.if_exists == "fail":
1038 raise ValueError(f"Table '{self.name}' already exists.")
1039 elif self.if_exists == "replace":
1040 self.pd_sql.drop_table(self.name, self.schema)
1041 self._execute_create()
1042 elif self.if_exists == "append":
1043 pass
1044 elif self.if_exists == "delete_rows":
1045 self.pd_sql.delete_rows(self.name, self.schema)
1046 else:
1047 raise ValueError(f"'{self.if_exists}' is not valid for if_exists")
1048 else:
1049 self._execute_create()
1051 def _execute_insert(self, conn, keys: list[str], data_iter) -> int:
1052 """
1053 Execute SQL statement inserting data
1055 Parameters
1056 ----------
1057 conn : sqlalchemy.engine.Engine or sqlalchemy.engine.Connection
1058 keys : list of str
1059 Column names
1060 data_iter : generator of list
1061 Each item contains a list of values to be inserted
1062 """
1063 data = [dict(zip(keys, row, strict=True)) for row in data_iter]
1064 result = self.pd_sql.execute(self.table.insert(), data)
1065 return result.rowcount
1067 def _execute_insert_multi(self, conn, keys: list[str], data_iter) -> int:
1068 """
1069 Alternative to _execute_insert for DBs support multi-value INSERT.
1071 Note: multi-value insert is usually faster for analytics DBs
1072 and tables containing a few columns
1073 but performance degrades quickly with increase of columns.
1075 """
1077 from sqlalchemy import insert
1079 data = [dict(zip(keys, row, strict=True)) for row in data_iter]
1080 stmt = insert(self.table).values(data)
1081 result = self.pd_sql.execute(stmt)
1082 return result.rowcount
1084 def insert_data(self) -> tuple[list[str], list[np.ndarray]]:
1085 if self.index is not None:
1086 temp = self.frame.copy(deep=False)
1087 temp.index.names = self.index
1088 try:
1089 temp = temp.reset_index()
1090 except ValueError as err:
1091 raise ValueError(f"duplicate name in index/columns: {err}") from err
1092 else:
1093 temp = self.frame
1095 column_names = list(map(str, temp.columns))
1096 ncols = len(column_names)
1097 # this just pre-allocates the list: None's will be replaced with ndarrays
1098 # error: List item 0 has incompatible type "None"; expected "ndarray"
1099 data_list: list[np.ndarray] = [None] * ncols # type: ignore[list-item]
1101 for i, (_, ser) in enumerate(temp.items()):
1102 if ser.dtype.kind == "M":
1103 if isinstance(ser._values, ArrowExtensionArray):
1104 import pyarrow as pa
1106 if pa.types.is_date(ser.dtype.pyarrow_dtype):
1107 # GH#53854 to_pydatetime not supported for pyarrow date dtypes
1108 d = ser._values.to_numpy(dtype=object)
1109 else:
1110 d = ser.dt.to_pydatetime()._values
1111 else:
1112 d = ser._values.to_pydatetime()
1113 elif ser.dtype.kind == "m":
1114 vals = ser._values
1115 if isinstance(vals, ArrowExtensionArray):
1116 vals = vals.to_numpy(dtype=np.dtype("m8[ns]"))
1117 # store as integers, see GH#6921, GH#7076
1118 d = vals.view("i8").astype(object)
1119 else:
1120 d = ser._values.astype(object)
1122 assert isinstance(d, np.ndarray), type(d)
1124 if ser._can_hold_na:
1125 # Note: this will miss timedeltas since they are converted to int
1126 mask = isna(d)
1127 d[mask] = None
1129 data_list[i] = d
1131 return column_names, data_list
1133 def insert(
1134 self,
1135 chunksize: int | None = None,
1136 method: Literal["multi"] | Callable | None = None,
1137 ) -> int | None:
1138 # set insert method
1139 if method is None:
1140 exec_insert = self._execute_insert
1141 elif method == "multi":
1142 exec_insert = self._execute_insert_multi
1143 elif callable(method):
1144 exec_insert = partial(method, self)
1145 else:
1146 raise ValueError(f"Invalid parameter `method`: {method}")
1148 keys, data_list = self.insert_data()
1150 nrows = len(self.frame)
1152 if nrows == 0:
1153 return 0
1155 if chunksize is None:
1156 chunksize = nrows
1157 elif chunksize == 0:
1158 raise ValueError("chunksize argument should be non-zero")
1160 chunks = (nrows // chunksize) + 1
1161 total_inserted = None
1162 with self.pd_sql.run_transaction() as conn:
1163 for i in range(chunks):
1164 start_i = i * chunksize
1165 end_i = min((i + 1) * chunksize, nrows)
1166 if start_i >= end_i:
1167 break
1169 chunk_iter = zip(
1170 *(arr[start_i:end_i] for arr in data_list), strict=True
1171 )
1172 num_inserted = exec_insert(conn, keys, chunk_iter)
1173 # GH 46891
1174 if num_inserted is not None:
1175 if total_inserted is None:
1176 total_inserted = num_inserted
1177 else:
1178 total_inserted += num_inserted
1179 return total_inserted
1181 def _query_iterator(
1182 self,
1183 result,
1184 exit_stack: ExitStack,
1185 chunksize: int | None,
1186 columns,
1187 coerce_float: bool = True,
1188 parse_dates=None,
1189 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1190 ) -> Generator[DataFrame]:
1191 """Return generator through chunked result set."""
1192 has_read_data = False
1193 with exit_stack:
1194 while True:
1195 data = result.fetchmany(chunksize)
1196 if not data:
1197 if not has_read_data:
1198 yield DataFrame.from_records(
1199 [], columns=columns, coerce_float=coerce_float
1200 )
1201 break
1203 has_read_data = True
1204 self.frame = _convert_arrays_to_dataframe(
1205 data, columns, coerce_float, dtype_backend
1206 )
1208 self._harmonize_columns(
1209 parse_dates=parse_dates, dtype_backend=dtype_backend
1210 )
1212 if self.index is not None:
1213 self.frame = self.frame.set_index(self.index)
1215 yield self.frame
1217 def read(
1218 self,
1219 exit_stack: ExitStack,
1220 coerce_float: bool = True,
1221 parse_dates=None,
1222 columns=None,
1223 chunksize: int | None = None,
1224 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1225 ) -> DataFrame | Iterator[DataFrame]:
1226 from sqlalchemy import select
1228 if columns is not None and len(columns) > 0:
1229 cols = [self.table.c[n] for n in columns]
1230 if self.index is not None:
1231 for idx in self.index[::-1]:
1232 cols.insert(0, self.table.c[idx])
1233 sql_select = select(*cols)
1234 else:
1235 sql_select = select(self.table)
1236 result = self.pd_sql.execute(sql_select)
1237 column_names = result.keys()
1239 if chunksize is not None:
1240 return self._query_iterator(
1241 result,
1242 exit_stack,
1243 chunksize,
1244 column_names,
1245 coerce_float=coerce_float,
1246 parse_dates=parse_dates,
1247 dtype_backend=dtype_backend,
1248 )
1249 else:
1250 data = result.fetchall()
1251 self.frame = _convert_arrays_to_dataframe(
1252 data, column_names, coerce_float, dtype_backend
1253 )
1255 self._harmonize_columns(
1256 parse_dates=parse_dates, dtype_backend=dtype_backend
1257 )
1259 if self.index is not None:
1260 self.frame = self.frame.set_index(self.index)
1262 return self.frame
1264 def _index_name(self, index, index_label):
1265 # for writing: index=True to include index in sql table
1266 if index is True:
1267 nlevels = self.frame.index.nlevels
1268 # if index_label is specified, set this as index name(s)
1269 if index_label is not None:
1270 if not isinstance(index_label, list):
1271 index_label = [index_label]
1272 if len(index_label) != nlevels:
1273 raise ValueError(
1274 "Length of 'index_label' should match number of "
1275 f"levels, which is {nlevels}"
1276 )
1277 return index_label
1278 # return the used column labels for the index columns
1279 if (
1280 nlevels == 1
1281 and "index" not in self.frame.columns
1282 and self.frame.index.name is None
1283 ):
1284 return ["index"]
1285 else:
1286 return com.fill_missing_names(self.frame.index.names)
1288 # for reading: index=(list of) string to specify column to set as index
1289 elif isinstance(index, str):
1290 return [index]
1291 elif isinstance(index, list):
1292 return index
1293 else:
1294 return None
1296 def _get_column_names_and_types(self, dtype_mapper):
1297 column_names_and_types = []
1298 if self.index is not None:
1299 for i, idx_label in enumerate(self.index):
1300 idx_type = dtype_mapper(self.frame.index._get_level_values(i))
1301 column_names_and_types.append((str(idx_label), idx_type, True))
1303 column_names_and_types += [
1304 (str(self.frame.columns[i]), dtype_mapper(self.frame.iloc[:, i]), False)
1305 for i in range(len(self.frame.columns))
1306 ]
1308 return column_names_and_types
1310 def _create_table_setup(self):
1311 from sqlalchemy import (
1312 Column,
1313 PrimaryKeyConstraint,
1314 Table,
1315 )
1316 from sqlalchemy.schema import MetaData
1318 column_names_and_types = self._get_column_names_and_types(self._sqlalchemy_type)
1320 columns: list[Any] = [
1321 Column(name, typ, index=is_index)
1322 for name, typ, is_index in column_names_and_types
1323 ]
1325 if self.keys is not None:
1326 if not is_list_like(self.keys):
1327 keys = [self.keys]
1328 else:
1329 keys = self.keys
1330 pkc = PrimaryKeyConstraint(*keys, name=self.name + "_pk")
1331 columns.append(pkc)
1333 schema = self.schema or self.pd_sql.meta.schema
1335 # At this point, attach to new metadata, only attach to self.meta
1336 # once table is created.
1337 meta = MetaData()
1338 return Table(self.name, meta, *columns, schema=schema)
1340 def _harmonize_columns(
1341 self,
1342 parse_dates=None,
1343 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1344 ) -> None:
1345 """
1346 Make the DataFrame's column types align with the SQL table
1347 column types.
1348 Need to work around limited NA value support. Floats are always
1349 fine, ints must always be floats if there are Null values.
1350 Booleans are hard because converting bool column with None replaces
1351 all Nones with false. Therefore only convert bool if there are no
1352 NA values.
1353 Datetimes should already be converted to np.datetime64 if supported,
1354 but here we also force conversion if required.
1355 """
1356 parse_dates = _process_parse_dates_argument(parse_dates)
1358 assert self.frame is not None # caller always sets frame before this
1359 for sql_col in self.table.columns:
1360 col_name = sql_col.name
1361 try:
1362 df_col = self.frame[col_name]
1364 # Handle date parsing upfront; don't try to convert columns
1365 # twice
1366 if col_name in parse_dates:
1367 try:
1368 fmt = parse_dates[col_name]
1369 except TypeError:
1370 fmt = None
1371 self.frame[col_name] = _handle_date_column(df_col, format=fmt)
1372 continue
1374 # the type the dataframe column should have
1375 col_type = self._get_dtype(sql_col.type)
1377 if (
1378 col_type is datetime
1379 or col_type is date
1380 or col_type is DatetimeTZDtype
1381 ):
1382 # Convert tz-aware Datetime SQL columns to UTC
1383 utc = col_type is DatetimeTZDtype
1384 self.frame[col_name] = _handle_date_column(df_col, utc=utc)
1385 elif dtype_backend == "numpy" and col_type is float:
1386 # floats support NA, can always convert!
1387 self.frame[col_name] = df_col.astype(col_type)
1388 elif (
1389 using_string_dtype()
1390 and is_string_dtype(col_type)
1391 and is_object_dtype(self.frame[col_name])
1392 ):
1393 self.frame[col_name] = df_col.astype(col_type)
1394 elif dtype_backend == "numpy" and len(df_col) == df_col.count():
1395 # No NA values, can convert ints and bools
1396 if col_type is np.dtype("int64") or col_type is bool:
1397 self.frame[col_name] = df_col.astype(col_type)
1398 except KeyError:
1399 pass # this column not in results
1401 def _sqlalchemy_type(self, col: Index | Series):
1402 dtype: DtypeArg = self.dtype or {}
1403 if is_dict_like(dtype):
1404 dtype = cast("dict", dtype)
1405 if col.name in dtype:
1406 return dtype[col.name]
1408 # Infer type of column, while ignoring missing values.
1409 # Needed for inserting typed data containing NULLs, GH 8778.
1410 col_type = lib.infer_dtype(col, skipna=True)
1412 from sqlalchemy.types import (
1413 TIMESTAMP,
1414 BigInteger,
1415 Boolean,
1416 Date,
1417 DateTime,
1418 Float,
1419 Integer,
1420 SmallInteger,
1421 Text,
1422 Time,
1423 )
1425 if col_type in ("datetime64", "datetime"):
1426 # GH 9086: TIMESTAMP is the suggested type if the column contains
1427 # timezone information
1428 try:
1429 # error: Item "Index" of "Union[Index, Series]" has no attribute "dt"
1430 if col.dt.tz is not None: # type: ignore[union-attr]
1431 return TIMESTAMP(timezone=True)
1432 except AttributeError:
1433 # The column is actually a DatetimeIndex
1434 # GH 26761 or an Index with date-like data e.g. 9999-01-01
1435 if getattr(col, "tz", None) is not None:
1436 return TIMESTAMP(timezone=True)
1437 return DateTime
1438 if col_type == "timedelta64":
1439 warnings.warn(
1440 "the 'timedelta' type is not supported, and will be "
1441 "written as integer values (ns frequency) to the database.",
1442 UserWarning,
1443 stacklevel=find_stack_level(),
1444 )
1445 return BigInteger
1446 elif col_type == "floating":
1447 if col.dtype == "float32":
1448 return Float(precision=23)
1449 else:
1450 return Float(precision=53)
1451 elif col_type == "integer":
1452 # GH35076 Map pandas integer to optimal SQLAlchemy integer type
1453 if col.dtype.name.lower() in ("int8", "uint8", "int16"):
1454 return SmallInteger
1455 elif col.dtype.name.lower() in ("uint16", "int32"):
1456 return Integer
1457 elif col.dtype.name.lower() == "uint64":
1458 raise ValueError("Unsigned 64 bit integer datatype is not supported")
1459 else:
1460 return BigInteger
1461 elif col_type == "boolean":
1462 return Boolean
1463 elif col_type == "date":
1464 return Date
1465 elif col_type == "time":
1466 return Time
1467 elif col_type == "complex":
1468 raise ValueError("Complex datatypes not supported")
1470 return Text
1472 def _get_dtype(self, sqltype):
1473 from sqlalchemy.types import (
1474 TIMESTAMP,
1475 Boolean,
1476 Date,
1477 DateTime,
1478 Float,
1479 Integer,
1480 String,
1481 )
1483 if isinstance(sqltype, Float):
1484 return float
1485 elif isinstance(sqltype, Integer):
1486 # TODO: Refine integer size.
1487 return np.dtype("int64")
1488 elif isinstance(sqltype, TIMESTAMP):
1489 # we have a timezone capable type
1490 if not sqltype.timezone:
1491 return datetime
1492 return DatetimeTZDtype
1493 elif isinstance(sqltype, DateTime):
1494 # Caution: np.datetime64 is also a subclass of np.number.
1495 return datetime
1496 elif isinstance(sqltype, Date):
1497 return date
1498 elif isinstance(sqltype, Boolean):
1499 return bool
1500 elif isinstance(sqltype, String):
1501 if using_string_dtype():
1502 return StringDtype(na_value=np.nan)
1504 return object
1507class PandasSQL(PandasObject, ABC):
1508 """
1509 Subclasses Should define read_query and to_sql.
1510 """
1512 def __enter__(self) -> Self:
1513 return self
1515 def __exit__(self, *args) -> None:
1516 pass
1518 def read_table(
1519 self,
1520 table_name: str,
1521 index_col: str | list[str] | None = None,
1522 coerce_float: bool = True,
1523 parse_dates=None,
1524 columns=None,
1525 schema: str | None = None,
1526 chunksize: int | None = None,
1527 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1528 ) -> DataFrame | Iterator[DataFrame]:
1529 raise NotImplementedError(
1530 "read_sql_table not supported for a DBAPI connection. Use "
1531 "read_sql_query, or pass a SQLAlchemy connectable."
1532 )
1534 @abstractmethod
1535 def read_query(
1536 self,
1537 sql: str,
1538 index_col: str | list[str] | None = None,
1539 coerce_float: bool = True,
1540 parse_dates=None,
1541 params=None,
1542 chunksize: int | None = None,
1543 dtype: DtypeArg | None = None,
1544 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1545 ) -> DataFrame | Iterator[DataFrame]:
1546 pass
1548 @abstractmethod
1549 def to_sql(
1550 self,
1551 frame,
1552 name: str,
1553 if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
1554 index: bool = True,
1555 index_label=None,
1556 schema=None,
1557 chunksize: int | None = None,
1558 dtype: DtypeArg | None = None,
1559 method: Literal["multi"] | Callable | None = None,
1560 engine: str = "auto",
1561 **engine_kwargs,
1562 ) -> int | None:
1563 pass
1565 @abstractmethod
1566 def execute(self, sql: str | Select | TextClause, params=None):
1567 pass
1569 @abstractmethod
1570 def has_table(self, name: str, schema: str | None = None) -> bool:
1571 pass
1573 @abstractmethod
1574 def _create_sql_schema(
1575 self,
1576 frame: DataFrame,
1577 table_name: str,
1578 keys: list[str] | None = None,
1579 dtype: DtypeArg | None = None,
1580 schema: str | None = None,
1581 ) -> str:
1582 pass
1585class BaseEngine:
1586 def insert_records(
1587 self,
1588 table: SQLTable,
1589 con,
1590 frame,
1591 name: str,
1592 index: bool | str | list[str] | None = True,
1593 schema=None,
1594 chunksize: int | None = None,
1595 method=None,
1596 **engine_kwargs,
1597 ) -> int | None:
1598 """
1599 Inserts data into already-prepared table
1600 """
1601 raise AbstractMethodError(self)
1604class SQLAlchemyEngine(BaseEngine):
1605 def __init__(self) -> None:
1606 import_optional_dependency(
1607 "sqlalchemy", extra="sqlalchemy is required for SQL support."
1608 )
1610 def insert_records(
1611 self,
1612 table: SQLTable,
1613 con,
1614 frame,
1615 name: str,
1616 index: bool | str | list[str] | None = True,
1617 schema=None,
1618 chunksize: int | None = None,
1619 method=None,
1620 **engine_kwargs,
1621 ) -> int | None:
1622 from sqlalchemy import exc
1624 try:
1625 return table.insert(chunksize=chunksize, method=method)
1626 except exc.StatementError as err:
1627 # GH34431
1628 # https://stackoverflow.com/a/67358288/6067848
1629 msg = r"""(\(1054, "Unknown column 'inf(e0)?' in 'field list'"\))(?#
1630 )|inf can not be used with MySQL"""
1631 err_text = str(err.orig)
1632 if re.search(msg, err_text):
1633 raise ValueError("inf cannot be used with MySQL") from err
1634 raise err
1637def get_engine(engine: str) -> BaseEngine:
1638 """return our implementation"""
1639 if engine == "auto":
1640 engine = config["io"]["sql"]["engine"]
1642 if engine == "auto":
1643 # try engines in this order
1644 engine_classes = [SQLAlchemyEngine]
1646 error_msgs = ""
1647 for engine_class in engine_classes:
1648 try:
1649 return engine_class()
1650 except ImportError as err:
1651 error_msgs += "\n - " + str(err)
1653 raise ImportError(
1654 "Unable to find a usable engine; "
1655 "tried using: 'sqlalchemy'.\n"
1656 "A suitable version of "
1657 "sqlalchemy is required for sql I/O "
1658 "support.\n"
1659 "Trying to import the above resulted in these errors:"
1660 f"{error_msgs}"
1661 )
1663 if engine == "sqlalchemy":
1664 return SQLAlchemyEngine()
1666 raise ValueError("engine must be one of 'auto', 'sqlalchemy'")
1669class SQLDatabase(PandasSQL):
1670 """
1671 This class enables conversion between DataFrame and SQL databases
1672 using SQLAlchemy to handle DataBase abstraction.
1674 Parameters
1675 ----------
1676 con : SQLAlchemy Connectable or URI string.
1677 Connectable to connect with the database. Using SQLAlchemy makes it
1678 possible to use any DB supported by that library.
1679 schema : string, default None
1680 Name of SQL schema in database to write to (if database flavor
1681 supports this). If None, use default schema (default).
1682 need_transaction : bool, default False
1683 If True, SQLDatabase will create a transaction.
1685 """
1687 def __init__(
1688 self, con, schema: str | None = None, need_transaction: bool = False
1689 ) -> None:
1690 from sqlalchemy import create_engine
1691 from sqlalchemy.engine import Engine
1692 from sqlalchemy.schema import MetaData
1694 # self.exit_stack cleans up the Engine and Connection and commits the
1695 # transaction if any of those objects was created below.
1696 # Cleanup happens either in self.__exit__ or at the end of the iterator
1697 # returned by read_sql when chunksize is not None.
1698 self.exit_stack = ExitStack()
1699 if isinstance(con, str):
1700 con = create_engine(con)
1701 self.exit_stack.callback(con.dispose)
1702 if isinstance(con, Engine):
1703 con = self.exit_stack.enter_context(con.connect())
1704 if need_transaction and not con.in_transaction():
1705 self.exit_stack.enter_context(con.begin())
1706 self.con = con
1707 self.meta = MetaData(schema=schema)
1708 self.returns_generator = False
1710 def __exit__(self, *args) -> None:
1711 if not self.returns_generator:
1712 self.exit_stack.close()
1714 @contextmanager
1715 def run_transaction(self):
1716 if not self.con.in_transaction():
1717 with self.con.begin():
1718 yield self.con
1719 else:
1720 yield self.con
1722 def execute(self, sql: str | Select | TextClause | Delete, params=None):
1723 """Simple passthrough to SQLAlchemy connectable"""
1724 from sqlalchemy.exc import SQLAlchemyError
1726 args = [] if params is None else [params]
1727 if isinstance(sql, str):
1728 execute_function = self.con.exec_driver_sql
1729 else:
1730 execute_function = self.con.execute
1732 try:
1733 return execute_function(sql, *args)
1734 except SQLAlchemyError as exc:
1735 raise DatabaseError(f"Execution failed on sql '{sql}': {exc}") from exc
1737 def read_table(
1738 self,
1739 table_name: str,
1740 index_col: str | list[str] | None = None,
1741 coerce_float: bool = True,
1742 parse_dates=None,
1743 columns=None,
1744 schema: str | None = None,
1745 chunksize: int | None = None,
1746 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1747 ) -> DataFrame | Iterator[DataFrame]:
1748 """
1749 Read SQL database table into a DataFrame.
1751 Parameters
1752 ----------
1753 table_name : str
1754 Name of SQL table in database.
1755 index_col : string, optional, default: None
1756 Column to set as index.
1757 coerce_float : bool, default True
1758 Attempts to convert values of non-string, non-numeric objects
1759 (like decimal.Decimal) to floating point. This can result in
1760 loss of precision.
1761 parse_dates : list or dict, default: None
1762 - List of column names to parse as dates.
1763 - Dict of ``{column_name: format string}`` where format string is
1764 strftime compatible in case of parsing string times, or is one of
1765 (D, s, ns, ms, us) in case of parsing integer timestamps.
1766 - Dict of ``{column_name: arg}``, where the arg corresponds
1767 to the keyword arguments of :func:`pandas.to_datetime`.
1768 Especially useful with databases without native Datetime support,
1769 such as SQLite.
1770 columns : list, default: None
1771 List of column names to select from SQL table.
1772 schema : string, default None
1773 Name of SQL schema in database to query (if database flavor
1774 supports this). If specified, this overwrites the default
1775 schema of the SQL database object.
1776 chunksize : int, default None
1777 If specified, return an iterator where `chunksize` is the number
1778 of rows to include in each chunk.
1779 dtype_backend : {'numpy_nullable', 'pyarrow'}
1780 Back-end data type applied to the resultant :class:`DataFrame`
1781 (still experimental). If not specified, the default behavior
1782 is to not use nullable data types. If specified, the behavior
1783 is as follows:
1785 * ``"numpy_nullable"``: returns nullable-dtype-backed :class:`DataFrame`
1786 * ``"pyarrow"``: returns pyarrow-backed nullable
1787 :class:`ArrowDtype` :class:`DataFrame`
1789 .. versionadded:: 2.0
1791 Returns
1792 -------
1793 DataFrame
1795 See Also
1796 --------
1797 pandas.read_sql_table
1798 SQLDatabase.read_query
1800 """
1801 self.meta.reflect(bind=self.con, only=[table_name], views=True)
1802 table = SQLTable(table_name, self, index=index_col, schema=schema)
1803 if chunksize is not None:
1804 self.returns_generator = True
1805 return table.read(
1806 self.exit_stack,
1807 coerce_float=coerce_float,
1808 parse_dates=parse_dates,
1809 columns=columns,
1810 chunksize=chunksize,
1811 dtype_backend=dtype_backend,
1812 )
1814 @staticmethod
1815 def _query_iterator(
1816 result,
1817 exit_stack: ExitStack,
1818 chunksize: int,
1819 columns,
1820 index_col=None,
1821 coerce_float: bool = True,
1822 parse_dates=None,
1823 dtype: DtypeArg | None = None,
1824 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1825 ) -> Generator[DataFrame]:
1826 """Return generator through chunked result set"""
1827 has_read_data = False
1828 with exit_stack:
1829 while True:
1830 data = result.fetchmany(chunksize)
1831 if not data:
1832 if not has_read_data:
1833 yield _wrap_result(
1834 [],
1835 columns,
1836 index_col=index_col,
1837 coerce_float=coerce_float,
1838 parse_dates=parse_dates,
1839 dtype=dtype,
1840 dtype_backend=dtype_backend,
1841 )
1842 break
1844 has_read_data = True
1845 yield _wrap_result(
1846 data,
1847 columns,
1848 index_col=index_col,
1849 coerce_float=coerce_float,
1850 parse_dates=parse_dates,
1851 dtype=dtype,
1852 dtype_backend=dtype_backend,
1853 )
1855 def read_query(
1856 self,
1857 sql: str,
1858 index_col: str | list[str] | None = None,
1859 coerce_float: bool = True,
1860 parse_dates=None,
1861 params=None,
1862 chunksize: int | None = None,
1863 dtype: DtypeArg | None = None,
1864 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
1865 ) -> DataFrame | Iterator[DataFrame]:
1866 """
1867 Read SQL query into a DataFrame.
1869 Parameters
1870 ----------
1871 sql : str
1872 SQL query to be executed.
1873 index_col : string, optional, default: None
1874 Column name to use as index for the returned DataFrame object.
1875 coerce_float : bool, default True
1876 Attempt to convert values of non-string, non-numeric objects (like
1877 decimal.Decimal) to floating point, useful for SQL result sets.
1878 params : list, tuple or dict, optional, default: None
1879 List of parameters to pass to execute method. The syntax used
1880 to pass parameters is database driver dependent. Check your
1881 database driver documentation for which of the five syntax styles,
1882 described in PEP 249's paramstyle, is supported.
1883 Eg. for psycopg2, uses %(name)s so use params={'name' : 'value'}
1884 parse_dates : list or dict, default: None
1885 - List of column names to parse as dates.
1886 - Dict of ``{column_name: format string}`` where format string is
1887 strftime compatible in case of parsing string times, or is one of
1888 (D, s, ns, ms, us) in case of parsing integer timestamps.
1889 - Dict of ``{column_name: arg dict}``, where the arg dict
1890 corresponds to the keyword arguments of
1891 :func:`pandas.to_datetime` Especially useful with databases
1892 without native Datetime support, such as SQLite.
1893 chunksize : int, default None
1894 If specified, return an iterator where `chunksize` is the number
1895 of rows to include in each chunk.
1896 dtype : Type name or dict of columns
1897 Data type for data or columns. E.g. np.float64 or
1898 {'a': np.float64, 'b': np.int32, 'c': 'Int64'}
1900 Returns
1901 -------
1902 DataFrame
1904 See Also
1905 --------
1906 read_sql_table : Read SQL database table into a DataFrame.
1907 read_sql
1909 """
1910 result = self.execute(sql, params)
1911 columns = result.keys()
1913 if chunksize is not None:
1914 self.returns_generator = True
1915 return self._query_iterator(
1916 result,
1917 self.exit_stack,
1918 chunksize,
1919 columns,
1920 index_col=index_col,
1921 coerce_float=coerce_float,
1922 parse_dates=parse_dates,
1923 dtype=dtype,
1924 dtype_backend=dtype_backend,
1925 )
1926 else:
1927 data = result.fetchall()
1928 frame = _wrap_result(
1929 data,
1930 columns,
1931 index_col=index_col,
1932 coerce_float=coerce_float,
1933 parse_dates=parse_dates,
1934 dtype=dtype,
1935 dtype_backend=dtype_backend,
1936 )
1937 return frame
1939 read_sql = read_query
1941 def prep_table(
1942 self,
1943 frame,
1944 name: str,
1945 if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
1946 index: bool | str | list[str] | None = True,
1947 index_label=None,
1948 schema=None,
1949 dtype: DtypeArg | None = None,
1950 ) -> SQLTable:
1951 """
1952 Prepares table in the database for data insertion. Creates it if needed, etc.
1953 """
1954 if dtype:
1955 if not is_dict_like(dtype):
1956 # error: Value expression in dictionary comprehension has incompatible
1957 # type "Union[ExtensionDtype, str, dtype[Any], Type[object],
1958 # Dict[Hashable, Union[ExtensionDtype, Union[str, dtype[Any]],
1959 # Type[str], Type[float], Type[int], Type[complex], Type[bool],
1960 # Type[object]]]]"; expected type "Union[ExtensionDtype, str,
1961 # dtype[Any], Type[object]]"
1962 dtype = dict.fromkeys(frame, dtype) # type: ignore[arg-type]
1963 else:
1964 dtype = cast("dict", dtype)
1966 from sqlalchemy.types import TypeEngine
1968 for col, my_type in dtype.items():
1969 if isinstance(my_type, type) and issubclass(my_type, TypeEngine):
1970 pass
1971 elif isinstance(my_type, TypeEngine):
1972 pass
1973 else:
1974 raise ValueError(f"The type of {col} is not a SQLAlchemy type")
1976 table = SQLTable(
1977 name,
1978 self,
1979 frame=frame,
1980 index=index,
1981 if_exists=if_exists,
1982 index_label=index_label,
1983 schema=schema,
1984 dtype=dtype,
1985 )
1986 table.create()
1987 return table
1989 def check_case_sensitive(
1990 self,
1991 name: str,
1992 schema: str | None,
1993 ) -> None:
1994 """
1995 Checks table name for issues with case-sensitivity.
1996 Method is called after data is inserted.
1997 """
1998 if not name.isdigit() and not name.islower():
1999 # check for potentially case sensitivity issues (GH7815)
2000 # Only check when name is not a number and name is not lower case
2001 from sqlalchemy import inspect as sqlalchemy_inspect
2003 insp = sqlalchemy_inspect(self.con)
2004 table_names = insp.get_table_names(schema=schema or self.meta.schema)
2005 if name not in table_names:
2006 msg = (
2007 f"The provided table name '{name}' is not found exactly as "
2008 "such in the database after writing the table, possibly "
2009 "due to case sensitivity issues. Consider using lower "
2010 "case table names."
2011 )
2012 warnings.warn(
2013 msg,
2014 UserWarning,
2015 stacklevel=find_stack_level(),
2016 )
2018 def to_sql(
2019 self,
2020 frame,
2021 name: str,
2022 if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
2023 index: bool = True,
2024 index_label=None,
2025 schema: str | None = None,
2026 chunksize: int | None = None,
2027 dtype: DtypeArg | None = None,
2028 method: Literal["multi"] | Callable | None = None,
2029 engine: str = "auto",
2030 **engine_kwargs,
2031 ) -> int | None:
2032 """
2033 Write records stored in a DataFrame to a SQL database.
2035 Parameters
2036 ----------
2037 frame : DataFrame
2038 name : string
2039 Name of SQL table.
2040 if_exists : {'fail', 'replace', 'append', 'delete_rows'}, default 'fail'
2041 - fail: If table exists, do nothing.
2042 - replace: If table exists, drop it, recreate it, and insert data.
2043 - append: If table exists, insert data. Create if does not exist.
2044 - delete_rows: If a table exists, delete all records and insert data.
2045 index : boolean, default True
2046 Write DataFrame index as a column.
2047 index_label : string or sequence, default None
2048 Column label for index column(s). If None is given (default) and
2049 `index` is True, then the index names are used.
2050 A sequence should be given if the DataFrame uses MultiIndex.
2051 schema : string, default None
2052 Name of SQL schema in database to write to (if database flavor
2053 supports this). If specified, this overwrites the default
2054 schema of the SQLDatabase object.
2055 chunksize : int, default None
2056 If not None, then rows will be written in batches of this size at a
2057 time. If None, all rows will be written at once.
2058 dtype : single type or dict of column name to SQL type, default None
2059 Optional specifying the datatype for columns. The SQL type should
2060 be a SQLAlchemy type. If all columns are of the same type, one
2061 single value can be used.
2062 method : {None', 'multi', callable}, default None
2063 Controls the SQL insertion clause used:
2065 * None : Uses standard SQL ``INSERT`` clause (one per row).
2066 * 'multi': Pass multiple values in a single ``INSERT`` clause.
2067 * callable with signature ``(pd_table, conn, keys, data_iter)``.
2069 Details and a sample callable implementation can be found in the
2070 section :ref:`insert method <io.sql.method>`.
2071 engine : {'auto', 'sqlalchemy'}, default 'auto'
2072 SQL engine library to use. If 'auto', then the option
2073 ``io.sql.engine`` is used. The default ``io.sql.engine``
2074 behavior is 'sqlalchemy'
2076 **engine_kwargs
2077 Any additional kwargs are passed to the engine.
2078 """
2079 sql_engine = get_engine(engine)
2081 table = self.prep_table(
2082 frame=frame,
2083 name=name,
2084 if_exists=if_exists,
2085 index=index,
2086 index_label=index_label,
2087 schema=schema,
2088 dtype=dtype,
2089 )
2091 total_inserted = sql_engine.insert_records(
2092 table=table,
2093 con=self.con,
2094 frame=frame,
2095 name=name,
2096 index=index,
2097 schema=schema,
2098 chunksize=chunksize,
2099 method=method,
2100 **engine_kwargs,
2101 )
2103 self.check_case_sensitive(name=name, schema=schema)
2104 return total_inserted
2106 @property
2107 def tables(self):
2108 return self.meta.tables
2110 def has_table(self, name: str, schema: str | None = None) -> bool:
2111 from sqlalchemy import inspect as sqlalchemy_inspect
2113 insp = sqlalchemy_inspect(self.con)
2114 return insp.has_table(name, schema or self.meta.schema)
2116 def get_table(self, table_name: str, schema: str | None = None) -> Table:
2117 from sqlalchemy import (
2118 Float,
2119 Numeric,
2120 Table,
2121 )
2123 schema = schema or self.meta.schema
2124 tbl = Table(table_name, self.meta, autoload_with=self.con, schema=schema)
2125 for column in tbl.columns:
2126 # Float is not a Numeric subclass in SQLAlchemy>=2.1
2127 if isinstance(column.type, (Numeric, Float)):
2128 column.type.asdecimal = False
2129 return tbl
2131 def drop_table(self, table_name: str, schema: str | None = None) -> None:
2132 schema = schema or self.meta.schema
2133 if self.has_table(table_name, schema):
2134 self.meta.reflect(
2135 bind=self.con, only=[table_name], schema=schema, views=True
2136 )
2137 with self.run_transaction():
2138 self.get_table(table_name, schema).drop(bind=self.con)
2139 self.meta.clear()
2141 def delete_rows(self, table_name: str, schema: str | None = None) -> None:
2142 schema = schema or self.meta.schema
2143 if self.has_table(table_name, schema):
2144 self.meta.reflect(
2145 bind=self.con, only=[table_name], schema=schema, views=True
2146 )
2147 table = self.get_table(table_name, schema)
2148 self.execute(table.delete()).close()
2149 self.meta.clear()
2151 def _create_sql_schema(
2152 self,
2153 frame: DataFrame,
2154 table_name: str,
2155 keys: list[str] | None = None,
2156 dtype: DtypeArg | None = None,
2157 schema: str | None = None,
2158 ) -> str:
2159 table = SQLTable(
2160 table_name,
2161 self,
2162 frame=frame,
2163 index=False,
2164 keys=keys,
2165 dtype=dtype,
2166 schema=schema,
2167 )
2168 return str(table.sql_schema())
2171# ---- SQL without SQLAlchemy ---
2174class ADBCDatabase(PandasSQL):
2175 """
2176 This class enables conversion between DataFrame and SQL databases
2177 using ADBC to handle DataBase abstraction.
2179 Parameters
2180 ----------
2181 con : adbc_driver_manager.dbapi.Connection
2182 """
2184 def __init__(self, con) -> None:
2185 self.con = con
2187 @contextmanager
2188 def run_transaction(self):
2189 with self.con.cursor() as cur:
2190 try:
2191 yield cur
2192 except Exception:
2193 self.con.rollback()
2194 raise
2195 self.con.commit()
2197 def execute(self, sql: str | Select | TextClause, params=None):
2198 from adbc_driver_manager import Error
2200 if not isinstance(sql, str):
2201 raise TypeError("Query must be a string unless using sqlalchemy.")
2202 args = [] if params is None else [params]
2203 cur = self.con.cursor()
2204 try:
2205 cur.execute(sql, *args)
2206 return cur
2207 except Error as exc:
2208 try:
2209 self.con.rollback()
2210 except Error as inner_exc: # pragma: no cover
2211 ex = DatabaseError(
2212 f"Execution failed on sql: {sql}\n{exc}\nunable to rollback"
2213 )
2214 raise ex from inner_exc
2216 ex = DatabaseError(f"Execution failed on sql '{sql}': {exc}")
2217 raise ex from exc
2219 def read_table(
2220 self,
2221 table_name: str,
2222 index_col: str | list[str] | None = None,
2223 coerce_float: bool = True,
2224 parse_dates=None,
2225 columns=None,
2226 schema: str | None = None,
2227 chunksize: int | None = None,
2228 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
2229 ) -> DataFrame | Iterator[DataFrame]:
2230 """
2231 Read SQL database table into a DataFrame.
2233 Parameters
2234 ----------
2235 table_name : str
2236 Name of SQL table in database.
2237 coerce_float : bool, default True
2238 Raises NotImplementedError
2239 parse_dates : list or dict, default: None
2240 - List of column names to parse as dates.
2241 - Dict of ``{column_name: format string}`` where format string is
2242 strftime compatible in case of parsing string times, or is one of
2243 (D, s, ns, ms, us) in case of parsing integer timestamps.
2244 - Dict of ``{column_name: arg}``, where the arg corresponds
2245 to the keyword arguments of :func:`pandas.to_datetime`.
2246 Especially useful with databases without native Datetime support,
2247 such as SQLite.
2248 columns : list, default: None
2249 List of column names to select from SQL table.
2250 schema : string, default None
2251 Name of SQL schema in database to query (if database flavor
2252 supports this). If specified, this overwrites the default
2253 schema of the SQL database object.
2254 chunksize : int, default None
2255 Raises NotImplementedError
2256 dtype_backend : {'numpy_nullable', 'pyarrow'}
2257 Back-end data type applied to the resultant :class:`DataFrame`
2258 (still experimental). If not specified, the default behavior
2259 is to not use nullable data types. If specified, the behavior
2260 is as follows:
2262 * ``"numpy_nullable"``: returns nullable-dtype-backed :class:`DataFrame`
2263 * ``"pyarrow"``: returns pyarrow-backed nullable
2264 :class:`ArrowDtype` :class:`DataFrame`
2266 .. versionadded:: 2.0
2268 Returns
2269 -------
2270 DataFrame
2272 See Also
2273 --------
2274 pandas.read_sql_table
2275 SQLDatabase.read_query
2277 """
2278 if coerce_float is not True:
2279 raise NotImplementedError(
2280 "'coerce_float' is not implemented for ADBC drivers"
2281 )
2282 if chunksize:
2283 raise NotImplementedError("'chunksize' is not implemented for ADBC drivers")
2285 if columns:
2286 if index_col:
2287 index_select = maybe_make_list(index_col)
2288 else:
2289 index_select = []
2290 to_select = index_select + columns
2291 select_list = ", ".join(_quote_identifier(x) for x in to_select)
2292 else:
2293 select_list = "*"
2294 quoted_table_name = _quote_identifier(table_name)
2295 if schema:
2296 quoted_schema = _quote_identifier(schema)
2297 stmt = f"SELECT {select_list} FROM {quoted_schema}.{quoted_table_name}"
2298 else:
2299 stmt = f"SELECT {select_list} FROM {quoted_table_name}"
2301 with self.execute(stmt) as cur:
2302 pa_table = cur.fetch_arrow_table()
2303 df = arrow_table_to_pandas(pa_table, dtype_backend=dtype_backend)
2305 return _wrap_result_adbc(
2306 df,
2307 index_col=index_col,
2308 parse_dates=parse_dates,
2309 )
2311 def read_query(
2312 self,
2313 sql: str,
2314 index_col: str | list[str] | None = None,
2315 coerce_float: bool = True,
2316 parse_dates=None,
2317 params=None,
2318 chunksize: int | None = None,
2319 dtype: DtypeArg | None = None,
2320 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
2321 ) -> DataFrame | Iterator[DataFrame]:
2322 """
2323 Read SQL query into a DataFrame.
2325 Parameters
2326 ----------
2327 sql : str
2328 SQL query to be executed.
2329 index_col : string, optional, default: None
2330 Column name to use as index for the returned DataFrame object.
2331 coerce_float : bool, default True
2332 Raises NotImplementedError
2333 params : list, tuple or dict, optional, default: None
2334 Raises NotImplementedError
2335 parse_dates : list or dict, default: None
2336 - List of column names to parse as dates.
2337 - Dict of ``{column_name: format string}`` where format string is
2338 strftime compatible in case of parsing string times, or is one of
2339 (D, s, ns, ms, us) in case of parsing integer timestamps.
2340 - Dict of ``{column_name: arg dict}``, where the arg dict
2341 corresponds to the keyword arguments of
2342 :func:`pandas.to_datetime` Especially useful with databases
2343 without native Datetime support, such as SQLite.
2344 chunksize : int, default None
2345 Raises NotImplementedError
2346 dtype : Type name or dict of columns
2347 Data type for data or columns. E.g. np.float64 or
2348 {'a': np.float64, 'b': np.int32, 'c': 'Int64'}
2350 Returns
2351 -------
2352 DataFrame
2354 See Also
2355 --------
2356 read_sql_table : Read SQL database table into a DataFrame.
2357 read_sql
2359 """
2360 if coerce_float is not True:
2361 raise NotImplementedError(
2362 "'coerce_float' is not implemented for ADBC drivers"
2363 )
2364 if params:
2365 raise NotImplementedError("'params' is not implemented for ADBC drivers")
2366 if chunksize:
2367 raise NotImplementedError("'chunksize' is not implemented for ADBC drivers")
2369 with self.execute(sql) as cur:
2370 pa_table = cur.fetch_arrow_table()
2371 df = arrow_table_to_pandas(pa_table, dtype_backend=dtype_backend)
2373 return _wrap_result_adbc(
2374 df,
2375 index_col=index_col,
2376 parse_dates=parse_dates,
2377 dtype=dtype,
2378 )
2380 read_sql = read_query
2382 def to_sql(
2383 self,
2384 frame,
2385 name: str,
2386 if_exists: Literal["fail", "replace", "append", "delete_rows"] = "fail",
2387 index: bool = True,
2388 index_label=None,
2389 schema: str | None = None,
2390 chunksize: int | None = None,
2391 dtype: DtypeArg | None = None,
2392 method: Literal["multi"] | Callable | None = None,
2393 engine: str = "auto",
2394 **engine_kwargs,
2395 ) -> int | None:
2396 """
2397 Write records stored in a DataFrame to a SQL database.
2399 Parameters
2400 ----------
2401 frame : DataFrame
2402 name : string
2403 Name of SQL table.
2404 if_exists : {'fail', 'replace', 'append'}, default 'fail'
2405 - fail: If table exists, do nothing.
2406 - replace: If table exists, drop it, recreate it, and insert data.
2407 - append: If table exists, insert data. Create if does not exist.
2408 - delete_rows: If a table exists, delete all records and insert data.
2409 index : boolean, default True
2410 Write DataFrame index as a column.
2411 index_label : string or sequence, default None
2412 Raises NotImplementedError
2413 schema : string, default None
2414 Name of SQL schema in database to write to (if database flavor
2415 supports this). If specified, this overwrites the default
2416 schema of the SQLDatabase object.
2417 chunksize : int, default None
2418 Raises NotImplementedError
2419 dtype : single type or dict of column name to SQL type, default None
2420 Raises NotImplementedError
2421 method : {None', 'multi', callable}, default None
2422 Raises NotImplementedError
2423 engine : {'auto', 'sqlalchemy'}, default 'auto'
2424 Raises NotImplementedError if not set to 'auto'
2425 """
2426 pa = import_optional_dependency("pyarrow")
2427 from adbc_driver_manager import Error
2429 if index_label:
2430 raise NotImplementedError(
2431 "'index_label' is not implemented for ADBC drivers"
2432 )
2433 if chunksize:
2434 raise NotImplementedError("'chunksize' is not implemented for ADBC drivers")
2435 if dtype:
2436 raise NotImplementedError("'dtype' is not implemented for ADBC drivers")
2437 if method:
2438 raise NotImplementedError("'method' is not implemented for ADBC drivers")
2439 if engine != "auto":
2440 raise NotImplementedError(
2441 "engine != 'auto' not implemented for ADBC drivers"
2442 )
2444 quoted_name = _quote_identifier(name)
2445 if schema:
2446 quoted_schema = _quote_identifier(schema)
2447 quoted_table_name = f"{quoted_schema}.{quoted_name}"
2448 else:
2449 quoted_table_name = quoted_name
2451 # pandas if_exists="append" will still create the
2452 # table if it does not exist; ADBC is more explicit with append/create
2453 # as applicable modes, so the semantics get blurred across
2454 # the libraries
2455 mode = "create"
2456 if self.has_table(name, schema):
2457 if if_exists == "fail":
2458 raise ValueError(f"Table '{name}' already exists.")
2459 elif if_exists == "replace":
2460 sql_statement = f"DROP TABLE {quoted_table_name}"
2461 self.execute(sql_statement).close()
2462 elif if_exists == "append":
2463 mode = "append"
2464 elif if_exists == "delete_rows":
2465 mode = "append"
2466 self.delete_rows(name, schema)
2468 try:
2469 with suppress_pyarrow_values_warning():
2470 tbl = pa.Table.from_pandas(frame, preserve_index=index)
2471 except pa.ArrowNotImplementedError as exc:
2472 raise ValueError("datatypes not supported") from exc
2474 with self.con.cursor() as cur:
2475 try:
2476 total_inserted = cur.adbc_ingest(
2477 table_name=name, data=tbl, mode=mode, db_schema_name=schema
2478 )
2479 except Error as exc:
2480 raise DatabaseError(
2481 f"Failed to insert records on table={name} with {mode=}"
2482 ) from exc
2484 self.con.commit()
2485 return total_inserted
2487 def has_table(self, name: str, schema: str | None = None) -> bool:
2488 meta = self.con.adbc_get_objects(
2489 depth="tables", db_schema_filter=schema, table_name_filter=name
2490 ).read_all()
2492 for catalog_schema in meta["catalog_db_schemas"].to_pylist():
2493 if not catalog_schema:
2494 continue
2495 for schema_record in catalog_schema:
2496 if not schema_record:
2497 continue
2499 for table_record in schema_record["db_schema_tables"]:
2500 if table_record["table_name"] == name:
2501 return True
2503 return False
2505 def delete_rows(self, name: str, schema: str | None = None) -> None:
2506 quoted_table_name = _quote_identifier(name)
2507 if schema:
2508 quoted_schema = _quote_identifier(schema)
2509 quoted_table_name = f"{quoted_schema}.{quoted_table_name}"
2511 if self.has_table(name, schema):
2512 self.execute(f"DELETE FROM {quoted_table_name}").close()
2514 def _create_sql_schema(
2515 self,
2516 frame: DataFrame,
2517 table_name: str,
2518 keys: list[str] | None = None,
2519 dtype: DtypeArg | None = None,
2520 schema: str | None = None,
2521 ) -> str:
2522 raise NotImplementedError("not implemented for adbc")
2525# sqlite-specific sql strings and handler class
2526# dictionary used for readability purposes
2527_SQL_TYPES = {
2528 "string": "TEXT",
2529 "floating": "REAL",
2530 "integer": "INTEGER",
2531 "datetime": "TIMESTAMP",
2532 "date": "DATE",
2533 "time": "TIME",
2534 "boolean": "INTEGER",
2535}
2538def _get_unicode_name(name: object) -> str:
2539 try:
2540 uname = str(name).encode("utf-8", "strict").decode("utf-8")
2541 except UnicodeError as err:
2542 raise ValueError(f"Cannot convert identifier to UTF-8: '{name}'") from err
2543 return uname
2546def _get_valid_sqlite_name(name: object) -> str:
2547 # See https://stackoverflow.com/questions/6514274/how-do-you-escape-strings\
2548 # -for-sqlite-table-column-names-in-python
2549 # Ensure the string can be encoded as UTF-8.
2550 # Ensure the string does not include any NUL characters.
2551 # Replace all " with "".
2552 # Wrap the entire thing in double quotes.
2554 uname = _get_unicode_name(name)
2555 if not len(uname):
2556 raise ValueError("Empty table or column name specified")
2558 nul_index = uname.find("\x00")
2559 if nul_index >= 0:
2560 raise ValueError("SQLite identifier cannot contain NULs")
2561 return '"' + uname.replace('"', '""') + '"'
2564def _quote_identifier(name: object) -> str:
2565 """
2566 Escape a SQL identifier (table name or schema name) for safe use in
2567 ADBC-generated SQL statements.
2569 Uses ANSI SQL double-quote escaping, which is supported by PostgreSQL,
2570 DuckDB, SQLite, and other databases targeted by ADBC drivers.
2572 Parameters
2573 ----------
2574 name : object
2575 Identifier to escape.
2577 Returns
2578 -------
2579 str
2580 The identifier wrapped in double quotes with any internal double
2581 quotes doubled, e.g. ``my"table`` becomes ``"my""table"``.
2583 Raises
2584 ------
2585 ValueError
2586 If *name* is empty or contains a NUL character.
2587 """
2588 uname = _get_unicode_name(name)
2589 if not len(uname):
2590 raise ValueError("Empty table or column name specified")
2592 nul_index = uname.find("\x00")
2593 if nul_index >= 0:
2594 raise ValueError("SQL identifier cannot contain NUL characters")
2596 return '"' + uname.replace('"', '""') + '"'
2599class SQLiteTable(SQLTable):
2600 """
2601 Patch the SQLTable for fallback support.
2602 Instead of a table variable just use the Create Table statement.
2603 """
2605 def __init__(self, *args, **kwargs) -> None:
2606 super().__init__(*args, **kwargs)
2608 self._register_date_adapters()
2610 def _register_date_adapters(self) -> None:
2611 # GH 8341
2612 # register an adapter callable for datetime.time object
2613 import sqlite3
2615 # this will transform time(12,34,56,789) into '12:34:56.000789'
2616 # (this is what sqlalchemy does)
2617 def _adapt_time(t) -> str:
2618 # This is faster than strftime
2619 return f"{t.hour:02d}:{t.minute:02d}:{t.second:02d}.{t.microsecond:06d}"
2621 # Also register adapters for date/datetime and co
2622 # xref https://docs.python.org/3.12/library/sqlite3.html#adapter-and-converter-recipes
2623 # Python 3.12+ doesn't auto-register adapters for us anymore
2625 adapt_date_iso = lambda val: val.isoformat()
2626 adapt_datetime_iso = lambda val: val.isoformat(" ")
2628 sqlite3.register_adapter(time, _adapt_time)
2630 sqlite3.register_adapter(date, adapt_date_iso)
2631 sqlite3.register_adapter(datetime, adapt_datetime_iso)
2633 convert_date = lambda val: date.fromisoformat(val.decode())
2634 convert_timestamp = lambda val: datetime.fromisoformat(val.decode())
2636 sqlite3.register_converter("date", convert_date)
2637 sqlite3.register_converter("timestamp", convert_timestamp)
2639 def sql_schema(self) -> str:
2640 return str(";\n".join(self.table))
2642 def _execute_create(self) -> None:
2643 with self.pd_sql.run_transaction() as cur:
2644 for stmt in self.table:
2645 cur.execute(stmt)
2647 def insert_statement(self, *, num_rows: int) -> str:
2648 names = list(map(str, self.frame.columns))
2649 wld = "?" # wildcard char
2650 escape = _get_valid_sqlite_name
2652 if self.index is not None:
2653 for idx in self.index[::-1]:
2654 names.insert(0, idx)
2656 bracketed_names = [escape(column) for column in names]
2657 col_names = ",".join(bracketed_names)
2659 row_wildcards = ",".join([wld] * len(names))
2660 wildcards = ",".join([f"({row_wildcards})" for _ in range(num_rows)])
2661 insert_statement = (
2662 f"INSERT INTO {escape(self.name)} ({col_names}) VALUES {wildcards}"
2663 )
2664 return insert_statement
2666 def _execute_insert(self, conn, keys, data_iter) -> int:
2667 from sqlite3 import Error
2669 data_list = list(data_iter)
2670 try:
2671 conn.executemany(self.insert_statement(num_rows=1), data_list)
2672 except Error as exc:
2673 raise DatabaseError("Execution failed") from exc
2674 return conn.rowcount
2676 def _execute_insert_multi(self, conn, keys, data_iter) -> int:
2677 data_list = list(data_iter)
2678 flattened_data = [x for row in data_list for x in row]
2679 conn.execute(self.insert_statement(num_rows=len(data_list)), flattened_data)
2680 return conn.rowcount
2682 def _create_table_setup(self):
2683 """
2684 Return a list of SQL statements that creates a table reflecting the
2685 structure of a DataFrame. The first entry will be a CREATE TABLE
2686 statement while the rest will be CREATE INDEX statements.
2687 """
2688 column_names_and_types = self._get_column_names_and_types(self._sql_type_name)
2689 escape = _get_valid_sqlite_name
2691 create_tbl_stmts = [
2692 escape(cname) + " " + ctype for cname, ctype, _ in column_names_and_types
2693 ]
2695 if self.keys is not None and len(self.keys):
2696 if not is_list_like(self.keys):
2697 keys = [self.keys]
2698 else:
2699 keys = self.keys
2700 cnames_br = ", ".join([escape(c) for c in keys])
2701 create_tbl_stmts.append(
2702 f"CONSTRAINT {self.name}_pk PRIMARY KEY ({cnames_br})"
2703 )
2704 if self.schema:
2705 schema_name = self.schema + "."
2706 else:
2707 schema_name = ""
2708 create_stmts = [
2709 "CREATE TABLE "
2710 + schema_name
2711 + escape(self.name)
2712 + " (\n"
2713 + ",\n ".join(create_tbl_stmts)
2714 + "\n)"
2715 ]
2717 ix_cols = [cname for cname, _, is_index in column_names_and_types if is_index]
2718 if ix_cols:
2719 cnames = "_".join(ix_cols)
2720 cnames_br = ",".join([escape(c) for c in ix_cols])
2721 create_stmts.append(
2722 "CREATE INDEX "
2723 + escape("ix_" + self.name + "_" + cnames)
2724 + "ON "
2725 + escape(self.name)
2726 + " ("
2727 + cnames_br
2728 + ")"
2729 )
2731 return create_stmts
2733 def _sql_type_name(self, col):
2734 dtype: DtypeArg = self.dtype or {}
2735 if is_dict_like(dtype):
2736 dtype = cast("dict", dtype)
2737 if col.name in dtype:
2738 return dtype[col.name]
2740 # Infer type of column, while ignoring missing values.
2741 # Needed for inserting typed data containing NULLs, GH 8778.
2742 col_type = lib.infer_dtype(col, skipna=True)
2744 if col_type == "timedelta64":
2745 warnings.warn(
2746 "the 'timedelta' type is not supported, and will be "
2747 "written as integer values (ns frequency) to the database.",
2748 UserWarning,
2749 stacklevel=find_stack_level(),
2750 )
2751 col_type = "integer"
2753 elif col_type == "datetime64":
2754 col_type = "datetime"
2756 elif col_type == "empty":
2757 col_type = "string"
2759 elif col_type == "complex":
2760 raise ValueError("Complex datatypes not supported")
2762 if col_type not in _SQL_TYPES:
2763 col_type = "string"
2765 return _SQL_TYPES[col_type]
2768class SQLiteDatabase(PandasSQL):
2769 """
2770 Version of SQLDatabase to support SQLite connections (fallback without
2771 SQLAlchemy). This should only be used internally.
2773 Parameters
2774 ----------
2775 con : sqlite connection object
2777 """
2779 def __init__(self, con) -> None:
2780 self.con = con
2782 @contextmanager
2783 def run_transaction(self):
2784 cur = self.con.cursor()
2785 try:
2786 yield cur
2787 self.con.commit()
2788 except Exception:
2789 self.con.rollback()
2790 raise
2791 finally:
2792 cur.close()
2794 def execute(self, sql: str | Select | TextClause, params=None):
2795 from sqlite3 import Error
2797 if not isinstance(sql, str):
2798 raise TypeError("Query must be a string unless using sqlalchemy.")
2799 args = [] if params is None else [params]
2800 cur = self.con.cursor()
2801 try:
2802 cur.execute(sql, *args)
2803 return cur
2804 except Error as exc:
2805 try:
2806 self.con.rollback()
2807 except Error as inner_exc: # pragma: no cover
2808 ex = DatabaseError(
2809 f"Execution failed on sql: {sql}\n{exc}\nunable to rollback"
2810 )
2811 raise ex from inner_exc
2813 if re.fullmatch(r"[A-Za-z_]\w*", sql.strip()):
2814 # GH#54233 a bare identifier is almost certainly an attempt to
2815 # read a table by name.
2816 ex = DatabaseError(
2817 f"Execution failed on sql '{sql}': {exc}\n\n"
2818 "Reading a table by name is only supported when using a "
2819 "SQLAlchemy connection. With a DBAPI connection, pass a SQL "
2820 f"query instead, e.g. 'SELECT * FROM {sql.strip()}'."
2821 )
2822 raise ex from exc
2824 ex = DatabaseError(f"Execution failed on sql '{sql}': {exc}")
2825 raise ex from exc
2827 @staticmethod
2828 def _query_iterator(
2829 cursor,
2830 chunksize: int,
2831 columns,
2832 index_col=None,
2833 coerce_float: bool = True,
2834 parse_dates=None,
2835 dtype: DtypeArg | None = None,
2836 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
2837 ) -> Generator[DataFrame]:
2838 """Return generator through chunked result set"""
2839 has_read_data = False
2840 while True:
2841 data = cursor.fetchmany(chunksize)
2842 if type(data) == tuple:
2843 data = list(data)
2844 if not data:
2845 cursor.close()
2846 if not has_read_data:
2847 result = DataFrame.from_records(
2848 [], columns=columns, coerce_float=coerce_float
2849 )
2850 if dtype:
2851 result = result.astype(dtype)
2852 yield result
2853 break
2855 has_read_data = True
2856 yield _wrap_result(
2857 data,
2858 columns,
2859 index_col=index_col,
2860 coerce_float=coerce_float,
2861 parse_dates=parse_dates,
2862 dtype=dtype,
2863 dtype_backend=dtype_backend,
2864 )
2866 def read_query(
2867 self,
2868 sql,
2869 index_col=None,
2870 coerce_float: bool = True,
2871 parse_dates=None,
2872 params=None,
2873 chunksize: int | None = None,
2874 dtype: DtypeArg | None = None,
2875 dtype_backend: DtypeBackend | Literal["numpy"] = "numpy",
2876 ) -> DataFrame | Iterator[DataFrame]:
2877 cursor = self.execute(sql, params)
2878 columns = [col_desc[0] for col_desc in cursor.description]
2880 if chunksize is not None:
2881 return self._query_iterator(
2882 cursor,
2883 chunksize,
2884 columns,
2885 index_col=index_col,
2886 coerce_float=coerce_float,
2887 parse_dates=parse_dates,
2888 dtype=dtype,
2889 dtype_backend=dtype_backend,
2890 )
2891 else:
2892 data = self._fetchall_as_list(cursor)
2893 cursor.close()
2895 frame = _wrap_result(
2896 data,
2897 columns,
2898 index_col=index_col,
2899 coerce_float=coerce_float,
2900 parse_dates=parse_dates,
2901 dtype=dtype,
2902 dtype_backend=dtype_backend,
2903 )
2904 return frame
2906 def _fetchall_as_list(self, cur):
2907 result = cur.fetchall()
2908 if not isinstance(result, list):
2909 result = list(result)
2910 return result
2912 def to_sql(
2913 self,
2914 frame,
2915 name: str,
2916 if_exists: str = "fail",
2917 index: bool = True,
2918 index_label=None,
2919 schema=None,
2920 chunksize: int | None = None,
2921 dtype: DtypeArg | None = None,
2922 method: Literal["multi"] | Callable | None = None,
2923 engine: str = "auto",
2924 **engine_kwargs,
2925 ) -> int | None:
2926 """
2927 Write records stored in a DataFrame to a SQL database.
2929 Parameters
2930 ----------
2931 frame: DataFrame
2932 name: string
2933 Name of SQL table.
2934 if_exists: {'fail', 'replace', 'append', 'delete_rows'}, default 'fail'
2935 fail: If table exists, do nothing.
2936 replace: If table exists, drop it, recreate it, and insert data.
2937 append: If table exists, insert data. Create if it does not exist.
2938 delete_rows: If a table exists, delete all records and insert data.
2939 index : bool, default True
2940 Write DataFrame index as a column
2941 index_label : string or sequence, default None
2942 Column label for index column(s). If None is given (default) and
2943 `index` is True, then the index names are used.
2944 A sequence should be given if the DataFrame uses MultiIndex.
2945 schema : string, default None
2946 Ignored parameter included for compatibility with SQLAlchemy
2947 version of ``to_sql``.
2948 chunksize : int, default None
2949 If not None, then rows will be written in batches of this
2950 size at a time. If None, all rows will be written at once.
2951 dtype : single type or dict of column name to SQL type, default None
2952 Optional specifying the datatype for columns. The SQL type should
2953 be a string. If all columns are of the same type, one single value
2954 can be used.
2955 method : {None, 'multi', callable}, default None
2956 Controls the SQL insertion clause used:
2958 * None : Uses standard SQL ``INSERT`` clause (one per row).
2959 * 'multi': Pass multiple values in a single ``INSERT`` clause.
2960 * callable with signature ``(pd_table, conn, keys, data_iter)``.
2962 Details and a sample callable implementation can be found in the
2963 section :ref:`insert method <io.sql.method>`.
2964 """
2965 if dtype:
2966 if not is_dict_like(dtype):
2967 # error: Value expression in dictionary comprehension has incompatible
2968 # type "Union[ExtensionDtype, str, dtype[Any], Type[object],
2969 # Dict[Hashable, Union[ExtensionDtype, Union[str, dtype[Any]],
2970 # Type[str], Type[float], Type[int], Type[complex], Type[bool],
2971 # Type[object]]]]"; expected type "Union[ExtensionDtype, str,
2972 # dtype[Any], Type[object]]"
2973 dtype = dict.fromkeys(frame, dtype) # type: ignore[arg-type]
2974 else:
2975 dtype = cast("dict", dtype)
2977 for col, my_type in dtype.items():
2978 if not isinstance(my_type, str):
2979 raise ValueError(
2980 f"Invalid type '{my_type}' for dtype of column '{col}': "
2981 "expected a string. When using a DB-API connection "
2982 "(e.g. sqlite3), dtype values must be SQL type "
2983 "strings (e.g. 'TEXT', 'FLOAT')."
2984 )
2986 table = SQLiteTable(
2987 name,
2988 self,
2989 frame=frame,
2990 index=index,
2991 if_exists=if_exists,
2992 index_label=index_label,
2993 dtype=dtype,
2994 )
2995 table.create()
2996 return table.insert(chunksize, method)
2998 def has_table(self, name: str, schema: str | None = None) -> bool:
2999 wld = "?"
3000 query = f"""
3001 SELECT
3002 name
3003 FROM
3004 sqlite_master
3005 WHERE
3006 type IN ('table', 'view')
3007 AND name={wld};
3008 """
3010 return len(self.execute(query, [name]).fetchall()) > 0
3012 def get_table(self, table_name: str, schema: str | None = None) -> None:
3013 return None # not supported in fallback mode
3015 def drop_table(self, name: str, schema: str | None = None) -> None:
3016 drop_sql = f"DROP TABLE {_get_valid_sqlite_name(name)}"
3017 self.execute(drop_sql).close()
3019 def delete_rows(self, name: str, schema: str | None = None) -> None:
3020 delete_sql = f"DELETE FROM {_get_valid_sqlite_name(name)}"
3021 if self.has_table(name, schema):
3022 self.execute(delete_sql).close()
3024 def _create_sql_schema(
3025 self,
3026 frame,
3027 table_name: str,
3028 keys=None,
3029 dtype: DtypeArg | None = None,
3030 schema: str | None = None,
3031 ) -> str:
3032 table = SQLiteTable(
3033 table_name,
3034 self,
3035 frame=frame,
3036 index=False,
3037 keys=keys,
3038 dtype=dtype,
3039 schema=schema,
3040 )
3041 return str(table.sql_schema())
3044def get_schema(
3045 frame,
3046 name: str,
3047 keys=None,
3048 con=None,
3049 dtype: DtypeArg | None = None,
3050 schema: str | None = None,
3051) -> str:
3052 """
3053 Get the SQL db table schema for the given frame.
3055 Parameters
3056 ----------
3057 frame : DataFrame
3058 name : str
3059 name of SQL table
3060 keys : string or sequence, default: None
3061 columns to use a primary key
3062 con: ADBC Connection, SQLAlchemy connectable, sqlite3 connection, default: None
3063 ADBC provides high performance I/O with native type support, where available.
3064 Using SQLAlchemy makes it possible to use any DB supported by that
3065 library
3066 If a DBAPI2 object, only sqlite3 is supported.
3067 dtype : dict of column name to SQL type, default None
3068 Optional specifying the datatype for columns. The SQL type should
3069 be a SQLAlchemy type, or a string for sqlite3 fallback connection.
3070 schema: str, default: None
3071 Optional specifying the schema to be used in creating the table.
3072 """
3073 with pandasSQL_builder(con=con) as pandas_sql:
3074 return pandas_sql._create_sql_schema(
3075 frame, name, keys=keys, dtype=dtype, schema=schema
3076 )