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# dialects/sqlite/base.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
10r'''
11.. dialect:: sqlite
12 :name: SQLite
13 :normal_support: 3.12+
14 :best_effort: 3.7.16+
16.. _sqlite_datetime:
18Date and Time Types
19-------------------
21SQLite does not have built-in DATE, TIME, or DATETIME types, and pysqlite does
22not provide out of the box functionality for translating values between Python
23`datetime` objects and a SQLite-supported format. SQLAlchemy's own
24:class:`~sqlalchemy.types.DateTime` and related types provide date formatting
25and parsing functionality when SQLite is used. The implementation classes are
26:class:`_sqlite.DATETIME`, :class:`_sqlite.DATE` and :class:`_sqlite.TIME`.
27These types represent dates and times as ISO formatted strings, which also
28nicely support ordering. There's no reliance on typical "libc" internals for
29these functions so historical dates are fully supported.
31Ensuring Text affinity
32^^^^^^^^^^^^^^^^^^^^^^
34The DDL rendered for these types is the standard ``DATE``, ``TIME``
35and ``DATETIME`` indicators. However, custom storage formats can also be
36applied to these types. When the
37storage format is detected as containing no alpha characters, the DDL for
38these types is rendered as ``DATE_CHAR``, ``TIME_CHAR``, and ``DATETIME_CHAR``,
39so that the column continues to have textual affinity.
41.. seealso::
43 `Type Affinity <https://www.sqlite.org/datatype3.html#affinity>`_ -
44 in the SQLite documentation
46.. _sqlite_autoincrement:
48SQLite Auto Incrementing Behavior
49----------------------------------
51Background on SQLite's autoincrement is at: https://sqlite.org/autoinc.html
53Key concepts:
55* SQLite has an implicit "auto increment" feature that takes place for any
56 non-composite primary-key column that is specifically created using
57 "INTEGER PRIMARY KEY" for the type + primary key.
59* SQLite also has an explicit "AUTOINCREMENT" keyword, that is **not**
60 equivalent to the implicit autoincrement feature; this keyword is not
61 recommended for general use. SQLAlchemy does not render this keyword
62 unless a special SQLite-specific directive is used (see below). However,
63 it still requires that the column's type is named "INTEGER".
65Using the AUTOINCREMENT Keyword
66^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
68To specifically render the AUTOINCREMENT keyword on the primary key column
69when rendering DDL, add the flag ``sqlite_autoincrement=True`` to the Table
70construct::
72 Table(
73 "sometable",
74 metadata,
75 Column("id", Integer, primary_key=True),
76 sqlite_autoincrement=True,
77 )
79Allowing autoincrement behavior SQLAlchemy types other than Integer/INTEGER
80^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
82SQLite's typing model is based on naming conventions. Among other things, this
83means that any type name which contains the substring ``"INT"`` will be
84determined to be of "integer affinity". A type named ``"BIGINT"``,
85``"SPECIAL_INT"`` or even ``"XYZINTQPR"``, will be considered by SQLite to be
86of "integer" affinity. However, **the SQLite autoincrement feature, whether
87implicitly or explicitly enabled, requires that the name of the column's type
88is exactly the string "INTEGER"**. Therefore, if an application uses a type
89like :class:`.BigInteger` for a primary key, on SQLite this type will need to
90be rendered as the name ``"INTEGER"`` when emitting the initial ``CREATE
91TABLE`` statement in order for the autoincrement behavior to be available.
93One approach to achieve this is to use :class:`.Integer` on SQLite
94only using :meth:`.TypeEngine.with_variant`::
96 table = Table(
97 "my_table",
98 metadata,
99 Column(
100 "id",
101 BigInteger().with_variant(Integer, "sqlite"),
102 primary_key=True,
103 ),
104 )
106Another is to use a subclass of :class:`.BigInteger` that overrides its DDL
107name to be ``INTEGER`` when compiled against SQLite::
109 from sqlalchemy import BigInteger
110 from sqlalchemy.ext.compiler import compiles
113 class SLBigInteger(BigInteger):
114 pass
117 @compiles(SLBigInteger, "sqlite")
118 def bi_c(element, compiler, **kw):
119 return "INTEGER"
122 @compiles(SLBigInteger)
123 def bi_c(element, compiler, **kw):
124 return compiler.visit_BIGINT(element, **kw)
127 table = Table(
128 "my_table", metadata, Column("id", SLBigInteger(), primary_key=True)
129 )
131.. seealso::
133 :meth:`.TypeEngine.with_variant`
135 :ref:`sqlalchemy.ext.compiler_toplevel`
137 `Datatypes In SQLite Version 3 <https://sqlite.org/datatype3.html>`_
139.. _sqlite_transactions:
141Transactions with SQLite and the sqlite3 driver
142-----------------------------------------------
144As a file-based database, SQLite's approach to transactions differs from
145traditional databases in many ways. Additionally, the ``sqlite3`` driver
146standard with Python (as well as the async version ``aiosqlite`` which builds
147on top of it) has several quirks, workarounds, and API features in the
148area of transaction control, all of which generally need to be addressed when
149constructing a SQLAlchemy application that uses SQLite.
151Legacy Transaction Mode with the sqlite3 driver
152^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
154The most important aspect of transaction handling with the sqlite3 driver is
155that it defaults (which will continue through Python 3.15 before being
156removed in Python 3.16) to legacy transactional behavior which does
157not strictly follow :pep:`249`. The way in which the driver diverges from the
158PEP is that it does not "begin" a transaction automatically as dictated by
159:pep:`249` except in the case of DML statements, e.g. INSERT, UPDATE, and
160DELETE. Normally, :pep:`249` dictates that a BEGIN must be emitted upon
161the first SQL statement of any kind, so that all subsequent operations will
162be established within a transaction until ``connection.commit()`` has been
163called. The ``sqlite3`` driver, in an effort to be easier to use in
164highly concurrent environments, skips this step for DQL (e.g. SELECT) statements,
165and also skips it for DDL (e.g. CREATE TABLE etc.) statements for more legacy
166reasons. Statements such as SAVEPOINT are also skipped.
168In modern versions of the ``sqlite3`` driver as of Python 3.12, this legacy
169mode of operation is referred to as
170`"legacy transaction control" <https://docs.python.org/3/library/sqlite3.html#sqlite3-transaction-control-isolation-level>`_, and is in
171effect by default due to the ``Connection.autocommit`` parameter being set to
172the constant ``sqlite3.LEGACY_TRANSACTION_CONTROL``. Prior to Python 3.12,
173the ``Connection.autocommit`` attribute did not exist.
175The implications of legacy transaction mode include:
177* **Incorrect support for transactional DDL** - statements like CREATE TABLE, ALTER TABLE,
178 CREATE INDEX etc. will not automatically BEGIN a transaction if one were not
179 started already, leading to the changes by each statement being
180 "autocommitted" immediately unless BEGIN were otherwise emitted first. Very
181 old (pre Python 3.6) versions of SQLite would also force a COMMIT for these
182 operations even if a transaction were present, however this is no longer the
183 case.
184* **SERIALIZABLE behavior not fully functional** - SQLite's transaction isolation
185 behavior is normally consistent with SERIALIZABLE isolation, as it is a file-
186 based system that locks the database file entirely for write operations,
187 preventing COMMIT until all reader transactions (and associated file locks)
188 have completed. However, sqlite3's legacy transaction mode fails to emit BEGIN for SELECT
189 statements, which causes these SELECT statements to no longer be "repeatable",
190 failing one of the consistency guarantees of SERIALIZABLE.
191* **Incorrect behavior for SAVEPOINT** - as the SAVEPOINT statement does not
192 imply a BEGIN, a new SAVEPOINT emitted before a BEGIN will function on its
193 own but fails to participate in the enclosing transaction, meaning a ROLLBACK
194 of the transaction will not rollback elements that were part of a released
195 savepoint.
197Legacy transaction mode first existed in order to faciliate working around
198SQLite's file locks. Because SQLite relies upon whole-file locks, it is easy to
199get "database is locked" errors, particularly when newer features like "write
200ahead logging" are disabled. This is a key reason why ``sqlite3``'s legacy
201transaction mode is still the default mode of operation; disabling it will
202produce behavior that is more susceptible to locked database errors. However
203note that **legacy transaction mode will no longer be the default** in a future
204Python version (3.16 as of this writing).
206.. _sqlite_enabling_transactions:
208Enabling Non-Legacy SQLite Transactional Modes with the sqlite3 or aiosqlite driver
209^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
211Current SQLAlchemy support allows either for setting the
212``.Connection.autocommit`` attribute, most directly by using a
213:func:`._sa.create_engine` parameter, or if on an older version of Python where
214the attribute is not available, using event hooks to control the behavior of
215BEGIN.
217* **Enabling modern sqlite3 transaction control via the autocommit connect parameter** (Python 3.12 and above)
219 To use SQLite in the mode described at `Transaction control via the autocommit attribute <https://docs.python.org/3/library/sqlite3.html#transaction-control-via-the-autocommit-attribute>`_,
220 the most straightforward approach is to set the attribute to its recommended value
221 of ``False`` at the connect level using :paramref:`_sa.create_engine.connect_args``::
223 from sqlalchemy import create_engine
225 engine = create_engine(
226 "sqlite:///myfile.db", connect_args={"autocommit": False}
227 )
229 This parameter is also passed through when using the aiosqlite driver::
231 from sqlalchemy.ext.asyncio import create_async_engine
233 engine = create_async_engine(
234 "sqlite+aiosqlite:///myfile.db", connect_args={"autocommit": False}
235 )
237 The parameter can also be set at the attribute level using the :meth:`.PoolEvents.connect`
238 event hook, however this will only work for sqlite3, as aiosqlite does not yet expose this
239 attribute on its ``Connection`` object::
241 from sqlalchemy import create_engine, event
243 engine = create_engine("sqlite:///myfile.db")
246 @event.listens_for(engine, "connect")
247 def do_connect(dbapi_connection, connection_record):
248 # enable autocommit=False mode
249 dbapi_connection.autocommit = False
251* **Using SQLAlchemy to emit BEGIN in lieu of SQLite's transaction control** (all Python versions, sqlite3 and aiosqlite)
253 For older versions of ``sqlite3`` or for cross-compatiblity with older and
254 newer versions, SQLAlchemy can also take over the job of transaction control.
255 This is achieved by using the :meth:`.ConnectionEvents.begin` hook
256 to emit the "BEGIN" command directly, while also disabling SQLite's control
257 of this command using the :meth:`.PoolEvents.connect` event hook to set the
258 ``Connection.isolation_level`` attribute to ``None``::
261 from sqlalchemy import create_engine, event
263 engine = create_engine("sqlite:///myfile.db")
266 @event.listens_for(engine, "connect")
267 def do_connect(dbapi_connection, connection_record):
268 # disable sqlite3's emitting of the BEGIN statement entirely.
269 dbapi_connection.isolation_level = None
272 @event.listens_for(engine, "begin")
273 def do_begin(conn):
274 # emit our own BEGIN. sqlite3 still emits COMMIT/ROLLBACK correctly
275 conn.exec_driver_sql("BEGIN")
277 When using the asyncio variant ``aiosqlite``, refer to ``engine.sync_engine``
278 as in the example below::
280 from sqlalchemy import create_engine, event
281 from sqlalchemy.ext.asyncio import create_async_engine
283 engine = create_async_engine("sqlite+aiosqlite:///myfile.db")
286 @event.listens_for(engine.sync_engine, "connect")
287 def do_connect(dbapi_connection, connection_record):
288 # disable aiosqlite's emitting of the BEGIN statement entirely.
289 dbapi_connection.isolation_level = None
292 @event.listens_for(engine.sync_engine, "begin")
293 def do_begin(conn):
294 # emit our own BEGIN. aiosqlite still emits COMMIT/ROLLBACK correctly
295 conn.exec_driver_sql("BEGIN")
297.. _sqlite_isolation_level:
299Using SQLAlchemy's Driver Level AUTOCOMMIT Feature with SQLite
300^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
302SQLAlchemy has a comprehensive database isolation feature with optional
303autocommit support that is introduced in the section :ref:`dbapi_autocommit`.
305For the ``sqlite3`` and ``aiosqlite`` drivers, SQLAlchemy only includes
306built-in support for "AUTOCOMMIT". Note that this mode is currently incompatible
307with the non-legacy isolation mode hooks documented in the previous
308section at :ref:`sqlite_enabling_transactions`.
310To use the ``sqlite3`` driver with SQLAlchemy driver-level autocommit,
311create an engine setting the :paramref:`_sa.create_engine.isolation_level`
312parameter to "AUTOCOMMIT"::
314 eng = create_engine("sqlite:///myfile.db", isolation_level="AUTOCOMMIT")
316When using the above mode, any event hooks that set the sqlite3 ``Connection.autocommit``
317parameter away from its default of ``sqlite3.LEGACY_TRANSACTION_CONTROL``
318as well as hooks that emit ``BEGIN`` should be disabled.
320Additional Reading for SQLite / sqlite3 transaction control
321^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
323Links with important information on SQLite, the sqlite3 driver,
324as well as long historical conversations on how things got to their current state:
326* `Isolation in SQLite <https://www.sqlite.org/isolation.html>`_ - on the SQLite website
327* `Transaction control <https://docs.python.org/3/library/sqlite3.html#transaction-control>`_ - describes the sqlite3 autocommit attribute as well
328 as the legacy isolation_level attribute.
329* `sqlite3 SELECT does not BEGIN a transaction, but should according to spec <https://github.com/python/cpython/issues/54133>`_ - imported Python standard library issue on github
330* `sqlite3 module breaks transactions and potentially corrupts data <https://github.com/python/cpython/issues/54949>`_ - imported Python standard library issue on github
333INSERT/UPDATE/DELETE...RETURNING
334---------------------------------
336The SQLite dialect supports SQLite 3.35's ``INSERT|UPDATE|DELETE..RETURNING``
337syntax. ``INSERT..RETURNING`` may be used
338automatically in some cases in order to fetch newly generated identifiers in
339place of the traditional approach of using ``cursor.lastrowid``, however
340``cursor.lastrowid`` is currently still preferred for simple single-statement
341cases for its better performance.
343To specify an explicit ``RETURNING`` clause, use the
344:meth:`._UpdateBase.returning` method on a per-statement basis::
346 # INSERT..RETURNING
347 result = connection.execute(
348 table.insert().values(name="foo").returning(table.c.col1, table.c.col2)
349 )
350 print(result.all())
352 # UPDATE..RETURNING
353 result = connection.execute(
354 table.update()
355 .where(table.c.name == "foo")
356 .values(name="bar")
357 .returning(table.c.col1, table.c.col2)
358 )
359 print(result.all())
361 # DELETE..RETURNING
362 result = connection.execute(
363 table.delete()
364 .where(table.c.name == "foo")
365 .returning(table.c.col1, table.c.col2)
366 )
367 print(result.all())
369.. versionadded:: 2.0 Added support for SQLite RETURNING
372.. _sqlite_foreign_keys:
374Foreign Key Support
375-------------------
377SQLite supports FOREIGN KEY syntax when emitting CREATE statements for tables,
378however by default these constraints have no effect on the operation of the
379table.
381Constraint checking on SQLite has three prerequisites:
383* At least version 3.6.19 of SQLite must be in use
384* The SQLite library must be compiled *without* the SQLITE_OMIT_FOREIGN_KEY
385 or SQLITE_OMIT_TRIGGER symbols enabled.
386* The ``PRAGMA foreign_keys = ON`` statement must be emitted on all
387 connections before use -- including the initial call to
388 :meth:`sqlalchemy.schema.MetaData.create_all`.
390SQLAlchemy allows for the ``PRAGMA`` statement to be emitted automatically for
391new connections through the usage of events::
393 from sqlalchemy.engine import Engine
394 from sqlalchemy import event
397 @event.listens_for(Engine, "connect")
398 def set_sqlite_pragma(dbapi_connection, connection_record):
399 # the sqlite3 driver will not set PRAGMA foreign_keys
400 # if autocommit=False; set to True temporarily
401 ac = dbapi_connection.autocommit
402 dbapi_connection.autocommit = True
404 cursor = dbapi_connection.cursor()
405 cursor.execute("PRAGMA foreign_keys=ON")
406 cursor.close()
408 # restore previous autocommit setting
409 dbapi_connection.autocommit = ac
411.. warning::
413 When SQLite foreign keys are enabled, it is **not possible**
414 to emit CREATE or DROP statements for tables that contain
415 mutually-dependent foreign key constraints;
416 to emit the DDL for these tables requires that ALTER TABLE be used to
417 create or drop these constraints separately, for which SQLite has
418 no support.
420.. seealso::
422 `SQLite Foreign Key Support <https://www.sqlite.org/foreignkeys.html>`_
423 - on the SQLite web site.
425 :ref:`event_toplevel` - SQLAlchemy event API.
427 :ref:`use_alter` - more information on SQLAlchemy's facilities for handling
428 mutually-dependent foreign key constraints.
430.. _sqlite_on_conflict_ddl:
432ON CONFLICT support for constraints
433-----------------------------------
435.. seealso:: This section describes the :term:`DDL` version of "ON CONFLICT" for
436 SQLite, which occurs within a CREATE TABLE statement. For "ON CONFLICT" as
437 applied to an INSERT statement, see :ref:`sqlite_on_conflict_insert`.
439SQLite supports a non-standard DDL clause known as ON CONFLICT which can be applied
440to primary key, unique, check, and not null constraints. In DDL, it is
441rendered either within the "CONSTRAINT" clause or within the column definition
442itself depending on the location of the target constraint. To render this
443clause within DDL, the extension parameter ``sqlite_on_conflict`` can be
444specified with a string conflict resolution algorithm within the
445:class:`.PrimaryKeyConstraint`, :class:`.UniqueConstraint`,
446:class:`.CheckConstraint` objects. Within the :class:`_schema.Column` object,
447there
448are individual parameters ``sqlite_on_conflict_not_null``,
449``sqlite_on_conflict_primary_key``, ``sqlite_on_conflict_unique`` which each
450correspond to the three types of relevant constraint types that can be
451indicated from a :class:`_schema.Column` object.
453.. seealso::
455 `ON CONFLICT <https://www.sqlite.org/lang_conflict.html>`_ - in the SQLite
456 documentation
458The ``sqlite_on_conflict`` parameters accept a string argument which is just
459the resolution name to be chosen, which on SQLite can be one of ROLLBACK,
460ABORT, FAIL, IGNORE, and REPLACE. For example, to add a UNIQUE constraint
461that specifies the IGNORE algorithm::
463 some_table = Table(
464 "some_table",
465 metadata,
466 Column("id", Integer, primary_key=True),
467 Column("data", Integer),
468 UniqueConstraint("id", "data", sqlite_on_conflict="IGNORE"),
469 )
471The above renders CREATE TABLE DDL as:
473.. sourcecode:: sql
475 CREATE TABLE some_table (
476 id INTEGER NOT NULL,
477 data INTEGER,
478 PRIMARY KEY (id),
479 UNIQUE (id, data) ON CONFLICT IGNORE
480 )
483When using the :paramref:`_schema.Column.unique`
484flag to add a UNIQUE constraint
485to a single column, the ``sqlite_on_conflict_unique`` parameter can
486be added to the :class:`_schema.Column` as well, which will be added to the
487UNIQUE constraint in the DDL::
489 some_table = Table(
490 "some_table",
491 metadata,
492 Column("id", Integer, primary_key=True),
493 Column(
494 "data", Integer, unique=True, sqlite_on_conflict_unique="IGNORE"
495 ),
496 )
498rendering:
500.. sourcecode:: sql
502 CREATE TABLE some_table (
503 id INTEGER NOT NULL,
504 data INTEGER,
505 PRIMARY KEY (id),
506 UNIQUE (data) ON CONFLICT IGNORE
507 )
509To apply the FAIL algorithm for a NOT NULL constraint,
510``sqlite_on_conflict_not_null`` is used::
512 some_table = Table(
513 "some_table",
514 metadata,
515 Column("id", Integer, primary_key=True),
516 Column(
517 "data", Integer, nullable=False, sqlite_on_conflict_not_null="FAIL"
518 ),
519 )
521this renders the column inline ON CONFLICT phrase:
523.. sourcecode:: sql
525 CREATE TABLE some_table (
526 id INTEGER NOT NULL,
527 data INTEGER NOT NULL ON CONFLICT FAIL,
528 PRIMARY KEY (id)
529 )
532Similarly, for an inline primary key, use ``sqlite_on_conflict_primary_key``::
534 some_table = Table(
535 "some_table",
536 metadata,
537 Column(
538 "id",
539 Integer,
540 primary_key=True,
541 sqlite_on_conflict_primary_key="FAIL",
542 ),
543 )
545SQLAlchemy renders the PRIMARY KEY constraint separately, so the conflict
546resolution algorithm is applied to the constraint itself:
548.. sourcecode:: sql
550 CREATE TABLE some_table (
551 id INTEGER NOT NULL,
552 PRIMARY KEY (id) ON CONFLICT FAIL
553 )
555.. _sqlite_on_conflict_insert:
557INSERT...ON CONFLICT (Upsert)
558-----------------------------
560.. seealso:: This section describes the :term:`DML` version of "ON CONFLICT" for
561 SQLite, which occurs within an INSERT statement. For "ON CONFLICT" as
562 applied to a CREATE TABLE statement, see :ref:`sqlite_on_conflict_ddl`.
564From version 3.24.0 onwards, SQLite supports "upserts" (update or insert)
565of rows into a table via the ``ON CONFLICT`` clause of the ``INSERT``
566statement. A candidate row will only be inserted if that row does not violate
567any unique or primary key constraints. In the case of a unique constraint violation, a
568secondary action can occur which can be either "DO UPDATE", indicating that
569the data in the target row should be updated, or "DO NOTHING", which indicates
570to silently skip this row.
572Conflicts are determined using columns that are part of existing unique
573constraints and indexes. These constraints are identified by stating the
574columns and conditions that comprise the indexes.
576SQLAlchemy provides ``ON CONFLICT`` support via the SQLite-specific
577:func:`_sqlite.insert()` function, which provides
578the generative methods :meth:`_sqlite.Insert.on_conflict_do_update`
579and :meth:`_sqlite.Insert.on_conflict_do_nothing`:
581.. sourcecode:: pycon+sql
583 >>> from sqlalchemy.dialects.sqlite import insert
585 >>> insert_stmt = insert(my_table).values(
586 ... id="some_existing_id", data="inserted value"
587 ... )
589 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
590 ... index_elements=["id"], set_=dict(data="updated value")
591 ... )
593 >>> print(do_update_stmt)
594 {printsql}INSERT INTO my_table (id, data) VALUES (?, ?)
595 ON CONFLICT (id) DO UPDATE SET data = ?{stop}
597 >>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing(index_elements=["id"])
599 >>> print(do_nothing_stmt)
600 {printsql}INSERT INTO my_table (id, data) VALUES (?, ?)
601 ON CONFLICT (id) DO NOTHING
603.. versionadded:: 1.4
605.. seealso::
607 `Upsert
608 <https://sqlite.org/lang_UPSERT.html>`_
609 - in the SQLite documentation.
612Specifying the Target
613^^^^^^^^^^^^^^^^^^^^^
615Both methods supply the "target" of the conflict using column inference:
617* The :paramref:`_sqlite.Insert.on_conflict_do_update.index_elements` argument
618 specifies a sequence containing string column names, :class:`_schema.Column`
619 objects, and/or SQL expression elements, which would identify a unique index
620 or unique constraint.
622* When using :paramref:`_sqlite.Insert.on_conflict_do_update.index_elements`
623 to infer an index, a partial index can be inferred by also specifying the
624 :paramref:`_sqlite.Insert.on_conflict_do_update.index_where` parameter:
626 .. sourcecode:: pycon+sql
628 >>> stmt = insert(my_table).values(user_email="a@b.com", data="inserted data")
630 >>> do_update_stmt = stmt.on_conflict_do_update(
631 ... index_elements=[my_table.c.user_email],
632 ... index_where=my_table.c.user_email.like("%@gmail.com"),
633 ... set_=dict(data=stmt.excluded.data),
634 ... )
636 >>> print(do_update_stmt)
637 {printsql}INSERT INTO my_table (data, user_email) VALUES (?, ?)
638 ON CONFLICT (user_email)
639 WHERE user_email LIKE '%@gmail.com'
640 DO UPDATE SET data = excluded.data
642The SET Clause
643^^^^^^^^^^^^^^^
645``ON CONFLICT...DO UPDATE`` is used to perform an update of the already
646existing row, using any combination of new values as well as values
647from the proposed insertion. These values are specified using the
648:paramref:`_sqlite.Insert.on_conflict_do_update.set_` parameter. This
649parameter accepts a dictionary which consists of direct values
650for UPDATE:
652.. sourcecode:: pycon+sql
654 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
656 >>> do_update_stmt = stmt.on_conflict_do_update(
657 ... index_elements=["id"], set_=dict(data="updated value")
658 ... )
660 >>> print(do_update_stmt)
661 {printsql}INSERT INTO my_table (id, data) VALUES (?, ?)
662 ON CONFLICT (id) DO UPDATE SET data = ?
664.. warning::
666 The :meth:`_sqlite.Insert.on_conflict_do_update` method does **not** take
667 into account Python-side default UPDATE values or generation functions,
668 e.g. those specified using :paramref:`_schema.Column.onupdate`. These
669 values will not be exercised for an ON CONFLICT style of UPDATE, unless
670 they are manually specified in the
671 :paramref:`_sqlite.Insert.on_conflict_do_update.set_` dictionary.
673Updating using the Excluded INSERT Values
674^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
676In order to refer to the proposed insertion row, the special alias
677:attr:`~.sqlite.Insert.excluded` is available as an attribute on
678the :class:`_sqlite.Insert` object; this object creates an "excluded." prefix
679on a column, that informs the DO UPDATE to update the row with the value that
680would have been inserted had the constraint not failed:
682.. sourcecode:: pycon+sql
684 >>> stmt = insert(my_table).values(
685 ... id="some_id", data="inserted value", author="jlh"
686 ... )
688 >>> do_update_stmt = stmt.on_conflict_do_update(
689 ... index_elements=["id"],
690 ... set_=dict(data="updated value", author=stmt.excluded.author),
691 ... )
693 >>> print(do_update_stmt)
694 {printsql}INSERT INTO my_table (id, data, author) VALUES (?, ?, ?)
695 ON CONFLICT (id) DO UPDATE SET data = ?, author = excluded.author
697Additional WHERE Criteria
698^^^^^^^^^^^^^^^^^^^^^^^^^
700The :meth:`_sqlite.Insert.on_conflict_do_update` method also accepts
701a WHERE clause using the :paramref:`_sqlite.Insert.on_conflict_do_update.where`
702parameter, which will limit those rows which receive an UPDATE:
704.. sourcecode:: pycon+sql
706 >>> stmt = insert(my_table).values(
707 ... id="some_id", data="inserted value", author="jlh"
708 ... )
710 >>> on_update_stmt = stmt.on_conflict_do_update(
711 ... index_elements=["id"],
712 ... set_=dict(data="updated value", author=stmt.excluded.author),
713 ... where=(my_table.c.status == 2),
714 ... )
715 >>> print(on_update_stmt)
716 {printsql}INSERT INTO my_table (id, data, author) VALUES (?, ?, ?)
717 ON CONFLICT (id) DO UPDATE SET data = ?, author = excluded.author
718 WHERE my_table.status = ?
721Skipping Rows with DO NOTHING
722^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
724``ON CONFLICT`` may be used to skip inserting a row entirely
725if any conflict with a unique constraint occurs; below this is illustrated
726using the :meth:`_sqlite.Insert.on_conflict_do_nothing` method:
728.. sourcecode:: pycon+sql
730 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
731 >>> stmt = stmt.on_conflict_do_nothing(index_elements=["id"])
732 >>> print(stmt)
733 {printsql}INSERT INTO my_table (id, data) VALUES (?, ?) ON CONFLICT (id) DO NOTHING
736If ``DO NOTHING`` is used without specifying any columns or constraint,
737it has the effect of skipping the INSERT for any unique violation which
738occurs:
740.. sourcecode:: pycon+sql
742 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
743 >>> stmt = stmt.on_conflict_do_nothing()
744 >>> print(stmt)
745 {printsql}INSERT INTO my_table (id, data) VALUES (?, ?) ON CONFLICT DO NOTHING
747.. _sqlite_type_reflection:
749Type Reflection
750---------------
752SQLite types are unlike those of most other database backends, in that
753the string name of the type usually does not correspond to a "type" in a
754one-to-one fashion. Instead, SQLite links per-column typing behavior
755to one of five so-called "type affinities" based on a string matching
756pattern for the type.
758SQLAlchemy's reflection process, when inspecting types, uses a simple
759lookup table to link the keywords returned to provided SQLAlchemy types.
760This lookup table is present within the SQLite dialect as it is for all
761other dialects. However, the SQLite dialect has a different "fallback"
762routine for when a particular type name is not located in the lookup map;
763it instead implements the SQLite "type affinity" scheme located at
764https://www.sqlite.org/datatype3.html section 2.1.
766The provided typemap will make direct associations from an exact string
767name match for the following types:
769:class:`_types.BIGINT`, :class:`_types.BLOB`,
770:class:`_types.BOOLEAN`, :class:`_types.BOOLEAN`,
771:class:`_types.CHAR`, :class:`_types.DATE`,
772:class:`_types.DATETIME`, :class:`_types.FLOAT`,
773:class:`_types.DECIMAL`, :class:`_types.FLOAT`,
774:class:`_types.INTEGER`, :class:`_types.INTEGER`,
775:class:`_types.NUMERIC`, :class:`_types.REAL`,
776:class:`_types.SMALLINT`, :class:`_types.TEXT`,
777:class:`_types.TIME`, :class:`_types.TIMESTAMP`,
778:class:`_types.VARCHAR`, :class:`_types.NVARCHAR`,
779:class:`_types.NCHAR`
781When a type name does not match one of the above types, the "type affinity"
782lookup is used instead:
784* :class:`_types.INTEGER` is returned if the type name includes the
785 string ``INT``
786* :class:`_types.TEXT` is returned if the type name includes the
787 string ``CHAR``, ``CLOB`` or ``TEXT``
788* :class:`_types.NullType` is returned if the type name includes the
789 string ``BLOB``
790* :class:`_types.REAL` is returned if the type name includes the string
791 ``REAL``, ``FLOA`` or ``DOUB``.
792* Otherwise, the :class:`_types.NUMERIC` type is used.
794.. _sqlite_partial_index:
796Partial Indexes
797---------------
799A partial index, e.g. one which uses a WHERE clause, can be specified
800with the DDL system using the argument ``sqlite_where``::
802 tbl = Table("testtbl", m, Column("data", Integer))
803 idx = Index(
804 "test_idx1",
805 tbl.c.data,
806 sqlite_where=and_(tbl.c.data > 5, tbl.c.data < 10),
807 )
809The index will be rendered at create time as:
811.. sourcecode:: sql
813 CREATE INDEX test_idx1 ON testtbl (data)
814 WHERE data > 5 AND data < 10
816.. _sqlite_dotted_column_names:
818Dotted Column Names
819-------------------
821Using table or column names that explicitly have periods in them is
822**not recommended**. While this is generally a bad idea for relational
823databases in general, as the dot is a syntactically significant character,
824the SQLite driver up until version **3.10.0** of SQLite has a bug which
825requires that SQLAlchemy filter out these dots in result sets.
827The bug, entirely outside of SQLAlchemy, can be illustrated thusly::
829 import sqlite3
831 assert sqlite3.sqlite_version_info < (
832 3,
833 10,
834 0,
835 ), "bug is fixed in this version"
837 conn = sqlite3.connect(":memory:")
838 cursor = conn.cursor()
840 cursor.execute("create table x (a integer, b integer)")
841 cursor.execute("insert into x (a, b) values (1, 1)")
842 cursor.execute("insert into x (a, b) values (2, 2)")
844 cursor.execute("select x.a, x.b from x")
845 assert [c[0] for c in cursor.description] == ["a", "b"]
847 cursor.execute(
848 """
849 select x.a, x.b from x where a=1
850 union
851 select x.a, x.b from x where a=2
852 """
853 )
854 assert [c[0] for c in cursor.description] == ["a", "b"], [
855 c[0] for c in cursor.description
856 ]
858The second assertion fails:
860.. sourcecode:: text
862 Traceback (most recent call last):
863 File "test.py", line 19, in <module>
864 [c[0] for c in cursor.description]
865 AssertionError: ['x.a', 'x.b']
867Where above, the driver incorrectly reports the names of the columns
868including the name of the table, which is entirely inconsistent vs.
869when the UNION is not present.
871SQLAlchemy relies upon column names being predictable in how they match
872to the original statement, so the SQLAlchemy dialect has no choice but
873to filter these out::
876 from sqlalchemy import create_engine
878 eng = create_engine("sqlite://")
879 conn = eng.connect()
881 conn.exec_driver_sql("create table x (a integer, b integer)")
882 conn.exec_driver_sql("insert into x (a, b) values (1, 1)")
883 conn.exec_driver_sql("insert into x (a, b) values (2, 2)")
885 result = conn.exec_driver_sql("select x.a, x.b from x")
886 assert result.keys() == ["a", "b"]
888 result = conn.exec_driver_sql(
889 """
890 select x.a, x.b from x where a=1
891 union
892 select x.a, x.b from x where a=2
893 """
894 )
895 assert result.keys() == ["a", "b"]
897Note that above, even though SQLAlchemy filters out the dots, *both
898names are still addressable*::
900 >>> row = result.first()
901 >>> row["a"]
902 1
903 >>> row["x.a"]
904 1
905 >>> row["b"]
906 1
907 >>> row["x.b"]
908 1
910Therefore, the workaround applied by SQLAlchemy only impacts
911:meth:`_engine.CursorResult.keys` and :meth:`.Row.keys()` in the public API. In
912the very specific case where an application is forced to use column names that
913contain dots, and the functionality of :meth:`_engine.CursorResult.keys` and
914:meth:`.Row.keys()` is required to return these dotted names unmodified,
915the ``sqlite_raw_colnames`` execution option may be provided, either on a
916per-:class:`_engine.Connection` basis::
918 result = conn.execution_options(sqlite_raw_colnames=True).exec_driver_sql(
919 """
920 select x.a, x.b from x where a=1
921 union
922 select x.a, x.b from x where a=2
923 """
924 )
925 assert result.keys() == ["x.a", "x.b"]
927or on a per-:class:`_engine.Engine` basis::
929 engine = create_engine(
930 "sqlite://", execution_options={"sqlite_raw_colnames": True}
931 )
933When using the per-:class:`_engine.Engine` execution option, note that
934**Core and ORM queries that use UNION may not function properly**.
936SQLite-specific table options
937-----------------------------
939One option for CREATE TABLE is supported directly by the SQLite
940dialect in conjunction with the :class:`_schema.Table` construct:
942* ``WITHOUT ROWID``::
944 Table("some_table", metadata, ..., sqlite_with_rowid=False)
946*
947 ``STRICT``::
949 Table("some_table", metadata, ..., sqlite_strict=True)
951 .. versionadded:: 2.0.37
953.. seealso::
955 `SQLite CREATE TABLE options
956 <https://www.sqlite.org/lang_createtable.html>`_
958.. _sqlite_include_internal:
960Reflecting internal schema tables
961----------------------------------
963Reflection methods that return lists of tables will omit so-called
964"SQLite internal schema object" names, which are considered by SQLite
965as any object name that is prefixed with ``sqlite_``. An example of
966such an object is the ``sqlite_sequence`` table that's generated when
967the ``AUTOINCREMENT`` column parameter is used. In order to return
968these objects, the parameter ``sqlite_include_internal=True`` may be
969passed to methods such as :meth:`_schema.MetaData.reflect` or
970:meth:`.Inspector.get_table_names`.
972.. versionadded:: 2.0 Added the ``sqlite_include_internal=True`` parameter.
973 Previously, these tables were not ignored by SQLAlchemy reflection
974 methods.
976.. note::
978 The ``sqlite_include_internal`` parameter does not refer to the
979 "system" tables that are present in schemas such as ``sqlite_master``.
981.. seealso::
983 `SQLite Internal Schema Objects <https://www.sqlite.org/fileformat2.html#intschema>`_ - in the SQLite
984 documentation.
986''' # noqa
987from __future__ import annotations
989import datetime
990import numbers
991import re
992from typing import Any
993from typing import Callable
994from typing import Optional
995from typing import TYPE_CHECKING
997from .json import JSON
998from .json import JSONIndexType
999from .json import JSONPathType
1000from ... import exc
1001from ... import schema as sa_schema
1002from ... import sql
1003from ... import text
1004from ... import types as sqltypes
1005from ... import util
1006from ...engine import default
1007from ...engine import processors
1008from ...engine import reflection
1009from ...engine.reflection import ReflectionDefaults
1010from ...sql import coercions
1011from ...sql import compiler
1012from ...sql import ddl as sa_ddl
1013from ...sql import elements
1014from ...sql import roles
1015from ...sql import schema
1016from ...types import BLOB # noqa
1017from ...types import BOOLEAN # noqa
1018from ...types import CHAR # noqa
1019from ...types import DECIMAL # noqa
1020from ...types import FLOAT # noqa
1021from ...types import INTEGER # noqa
1022from ...types import NUMERIC # noqa
1023from ...types import REAL # noqa
1024from ...types import SMALLINT # noqa
1025from ...types import TEXT # noqa
1026from ...types import TIMESTAMP # noqa
1027from ...types import VARCHAR # noqa
1029if TYPE_CHECKING:
1030 from ...engine.interfaces import DBAPIConnection
1031 from ...engine.interfaces import Dialect
1032 from ...engine.interfaces import IsolationLevel
1033 from ...sql.type_api import _BindProcessorType
1034 from ...sql.type_api import _ResultProcessorType
1037class _SQliteJson(JSON):
1038 def result_processor(self, dialect, coltype):
1039 default_processor = super().result_processor(dialect, coltype)
1041 def process(value):
1042 try:
1043 return default_processor(value)
1044 except TypeError:
1045 if isinstance(value, numbers.Number):
1046 return value
1047 else:
1048 raise
1050 return process
1053class _DateTimeMixin:
1054 _reg = None
1055 _storage_format = None
1057 def __init__(self, storage_format=None, regexp=None, **kw):
1058 super().__init__(**kw)
1059 if regexp is not None:
1060 self._reg = re.compile(regexp)
1061 if storage_format is not None:
1062 self._storage_format = storage_format
1064 @property
1065 def format_is_text_affinity(self):
1066 """return True if the storage format will automatically imply
1067 a TEXT affinity.
1069 If the storage format contains no non-numeric characters,
1070 it will imply a NUMERIC storage format on SQLite; in this case,
1071 the type will generate its DDL as DATE_CHAR, DATETIME_CHAR,
1072 TIME_CHAR.
1074 """
1075 spec = self._storage_format % {
1076 "year": 0,
1077 "month": 0,
1078 "day": 0,
1079 "hour": 0,
1080 "minute": 0,
1081 "second": 0,
1082 "microsecond": 0,
1083 }
1084 return bool(re.search(r"[^0-9]", spec))
1086 def adapt(self, cls, **kw):
1087 if issubclass(cls, _DateTimeMixin):
1088 if self._storage_format:
1089 kw["storage_format"] = self._storage_format
1090 if self._reg:
1091 kw["regexp"] = self._reg
1092 return super().adapt(cls, **kw)
1094 def literal_processor(self, dialect):
1095 bp = self.bind_processor(dialect)
1097 def process(value):
1098 return "'%s'" % bp(value)
1100 return process
1103class DATETIME(_DateTimeMixin, sqltypes.DateTime):
1104 r"""Represent a Python datetime object in SQLite using a string.
1106 The default string storage format is::
1108 "%(year)04d-%(month)02d-%(day)02d %(hour)02d:%(minute)02d:%(second)02d.%(microsecond)06d"
1110 e.g.:
1112 .. sourcecode:: text
1114 2021-03-15 12:05:57.105542
1116 The incoming storage format is by default parsed using the
1117 Python ``datetime.fromisoformat()`` function.
1119 .. versionchanged:: 2.0 ``datetime.fromisoformat()`` is used for default
1120 datetime string parsing.
1122 The storage format can be customized to some degree using the
1123 ``storage_format`` and ``regexp`` parameters, such as::
1125 import re
1126 from sqlalchemy.dialects.sqlite import DATETIME
1128 dt = DATETIME(
1129 storage_format=(
1130 "%(year)04d/%(month)02d/%(day)02d %(hour)02d:%(minute)02d:%(second)02d"
1131 ),
1132 regexp=r"(\d+)/(\d+)/(\d+) (\d+)-(\d+)-(\d+)",
1133 )
1135 :param truncate_microseconds: when ``True`` microseconds will be truncated
1136 from the datetime. Can't be specified together with ``storage_format``
1137 or ``regexp``.
1139 :param storage_format: format string which will be applied to the dict
1140 with keys year, month, day, hour, minute, second, and microsecond.
1142 :param regexp: regular expression which will be applied to incoming result
1143 rows, replacing the use of ``datetime.fromisoformat()`` to parse incoming
1144 strings. If the regexp contains named groups, the resulting match dict is
1145 applied to the Python datetime() constructor as keyword arguments.
1146 Otherwise, if positional groups are used, the datetime() constructor
1147 is called with positional arguments via
1148 ``*map(int, match_obj.groups(0))``.
1150 """ # noqa
1152 _storage_format = (
1153 "%(year)04d-%(month)02d-%(day)02d "
1154 "%(hour)02d:%(minute)02d:%(second)02d.%(microsecond)06d"
1155 )
1157 def __init__(self, *args, **kwargs):
1158 truncate_microseconds = kwargs.pop("truncate_microseconds", False)
1159 super().__init__(*args, **kwargs)
1160 if truncate_microseconds:
1161 assert "storage_format" not in kwargs, (
1162 "You can specify only "
1163 "one of truncate_microseconds or storage_format."
1164 )
1165 assert "regexp" not in kwargs, (
1166 "You can specify only one of "
1167 "truncate_microseconds or regexp."
1168 )
1169 self._storage_format = (
1170 "%(year)04d-%(month)02d-%(day)02d "
1171 "%(hour)02d:%(minute)02d:%(second)02d"
1172 )
1174 def bind_processor(
1175 self, dialect: Dialect
1176 ) -> Optional[_BindProcessorType[Any]]:
1177 datetime_datetime = datetime.datetime
1178 datetime_date = datetime.date
1179 format_ = self._storage_format
1181 def process(value):
1182 if value is None:
1183 return None
1184 elif isinstance(value, datetime_datetime):
1185 return format_ % {
1186 "year": value.year,
1187 "month": value.month,
1188 "day": value.day,
1189 "hour": value.hour,
1190 "minute": value.minute,
1191 "second": value.second,
1192 "microsecond": value.microsecond,
1193 }
1194 elif isinstance(value, datetime_date):
1195 return format_ % {
1196 "year": value.year,
1197 "month": value.month,
1198 "day": value.day,
1199 "hour": 0,
1200 "minute": 0,
1201 "second": 0,
1202 "microsecond": 0,
1203 }
1204 else:
1205 raise TypeError(
1206 "SQLite DateTime type only accepts Python "
1207 "datetime and date objects as input."
1208 )
1210 return process
1212 def result_processor(
1213 self, dialect: Dialect, coltype: object
1214 ) -> Optional[_ResultProcessorType[Any]]:
1215 if self._reg:
1216 return processors.str_to_datetime_processor_factory(
1217 self._reg, datetime.datetime
1218 )
1219 else:
1220 return processors.str_to_datetime
1223class DATE(_DateTimeMixin, sqltypes.Date):
1224 r"""Represent a Python date object in SQLite using a string.
1226 The default string storage format is::
1228 "%(year)04d-%(month)02d-%(day)02d"
1230 e.g.:
1232 .. sourcecode:: text
1234 2011-03-15
1236 The incoming storage format is by default parsed using the
1237 Python ``date.fromisoformat()`` function.
1239 .. versionchanged:: 2.0 ``date.fromisoformat()`` is used for default
1240 date string parsing.
1243 The storage format can be customized to some degree using the
1244 ``storage_format`` and ``regexp`` parameters, such as::
1246 import re
1247 from sqlalchemy.dialects.sqlite import DATE
1249 d = DATE(
1250 storage_format="%(month)02d/%(day)02d/%(year)04d",
1251 regexp=re.compile("(?P<month>\d+)/(?P<day>\d+)/(?P<year>\d+)"),
1252 )
1254 :param storage_format: format string which will be applied to the
1255 dict with keys year, month, and day.
1257 :param regexp: regular expression which will be applied to
1258 incoming result rows, replacing the use of ``date.fromisoformat()`` to
1259 parse incoming strings. If the regexp contains named groups, the resulting
1260 match dict is applied to the Python date() constructor as keyword
1261 arguments. Otherwise, if positional groups are used, the date()
1262 constructor is called with positional arguments via
1263 ``*map(int, match_obj.groups(0))``.
1265 """
1267 _storage_format = "%(year)04d-%(month)02d-%(day)02d"
1269 def bind_processor(
1270 self, dialect: Dialect
1271 ) -> Optional[_BindProcessorType[Any]]:
1272 datetime_date = datetime.date
1273 format_ = self._storage_format
1275 def process(value):
1276 if value is None:
1277 return None
1278 elif isinstance(value, datetime_date):
1279 return format_ % {
1280 "year": value.year,
1281 "month": value.month,
1282 "day": value.day,
1283 }
1284 else:
1285 raise TypeError(
1286 "SQLite Date type only accepts Python "
1287 "date objects as input."
1288 )
1290 return process
1292 def result_processor(
1293 self, dialect: Dialect, coltype: object
1294 ) -> Optional[_ResultProcessorType[Any]]:
1295 if self._reg:
1296 return processors.str_to_datetime_processor_factory(
1297 self._reg, datetime.date
1298 )
1299 else:
1300 return processors.str_to_date
1303class TIME(_DateTimeMixin, sqltypes.Time):
1304 r"""Represent a Python time object in SQLite using a string.
1306 The default string storage format is::
1308 "%(hour)02d:%(minute)02d:%(second)02d.%(microsecond)06d"
1310 e.g.:
1312 .. sourcecode:: text
1314 12:05:57.10558
1316 The incoming storage format is by default parsed using the
1317 Python ``time.fromisoformat()`` function.
1319 .. versionchanged:: 2.0 ``time.fromisoformat()`` is used for default
1320 time string parsing.
1322 The storage format can be customized to some degree using the
1323 ``storage_format`` and ``regexp`` parameters, such as::
1325 import re
1326 from sqlalchemy.dialects.sqlite import TIME
1328 t = TIME(
1329 storage_format="%(hour)02d-%(minute)02d-%(second)02d-%(microsecond)06d",
1330 regexp=re.compile("(\d+)-(\d+)-(\d+)-(?:-(\d+))?"),
1331 )
1333 :param truncate_microseconds: when ``True`` microseconds will be truncated
1334 from the time. Can't be specified together with ``storage_format``
1335 or ``regexp``.
1337 :param storage_format: format string which will be applied to the dict
1338 with keys hour, minute, second, and microsecond.
1340 :param regexp: regular expression which will be applied to incoming result
1341 rows, replacing the use of ``datetime.fromisoformat()`` to parse incoming
1342 strings. If the regexp contains named groups, the resulting match dict is
1343 applied to the Python time() constructor as keyword arguments. Otherwise,
1344 if positional groups are used, the time() constructor is called with
1345 positional arguments via ``*map(int, match_obj.groups(0))``.
1347 """
1349 _storage_format = "%(hour)02d:%(minute)02d:%(second)02d.%(microsecond)06d"
1351 def __init__(self, *args, **kwargs):
1352 truncate_microseconds = kwargs.pop("truncate_microseconds", False)
1353 super().__init__(*args, **kwargs)
1354 if truncate_microseconds:
1355 assert "storage_format" not in kwargs, (
1356 "You can specify only "
1357 "one of truncate_microseconds or storage_format."
1358 )
1359 assert "regexp" not in kwargs, (
1360 "You can specify only one of "
1361 "truncate_microseconds or regexp."
1362 )
1363 self._storage_format = "%(hour)02d:%(minute)02d:%(second)02d"
1365 def bind_processor(self, dialect):
1366 datetime_time = datetime.time
1367 format_ = self._storage_format
1369 def process(value):
1370 if value is None:
1371 return None
1372 elif isinstance(value, datetime_time):
1373 return format_ % {
1374 "hour": value.hour,
1375 "minute": value.minute,
1376 "second": value.second,
1377 "microsecond": value.microsecond,
1378 }
1379 else:
1380 raise TypeError(
1381 "SQLite Time type only accepts Python "
1382 "time objects as input."
1383 )
1385 return process
1387 def result_processor(self, dialect, coltype):
1388 if self._reg:
1389 return processors.str_to_datetime_processor_factory(
1390 self._reg, datetime.time
1391 )
1392 else:
1393 return processors.str_to_time
1396colspecs = {
1397 sqltypes.Date: DATE,
1398 sqltypes.DateTime: DATETIME,
1399 sqltypes.JSON: _SQliteJson,
1400 sqltypes.JSON.JSONIndexType: JSONIndexType,
1401 sqltypes.JSON.JSONPathType: JSONPathType,
1402 sqltypes.Time: TIME,
1403}
1405ischema_names = {
1406 "BIGINT": sqltypes.BIGINT,
1407 "BLOB": sqltypes.BLOB,
1408 "BOOL": sqltypes.BOOLEAN,
1409 "BOOLEAN": sqltypes.BOOLEAN,
1410 "CHAR": sqltypes.CHAR,
1411 "DATE": sqltypes.DATE,
1412 "DATE_CHAR": sqltypes.DATE,
1413 "DATETIME": sqltypes.DATETIME,
1414 "DATETIME_CHAR": sqltypes.DATETIME,
1415 "DOUBLE": sqltypes.DOUBLE,
1416 "DECIMAL": sqltypes.DECIMAL,
1417 "FLOAT": sqltypes.FLOAT,
1418 "INT": sqltypes.INTEGER,
1419 "INTEGER": sqltypes.INTEGER,
1420 "JSON": JSON,
1421 "NUMERIC": sqltypes.NUMERIC,
1422 "REAL": sqltypes.REAL,
1423 "SMALLINT": sqltypes.SMALLINT,
1424 "TEXT": sqltypes.TEXT,
1425 "TIME": sqltypes.TIME,
1426 "TIME_CHAR": sqltypes.TIME,
1427 "TIMESTAMP": sqltypes.TIMESTAMP,
1428 "VARCHAR": sqltypes.VARCHAR,
1429 "NVARCHAR": sqltypes.NVARCHAR,
1430 "NCHAR": sqltypes.NCHAR,
1431}
1434class SQLiteCompiler(compiler.SQLCompiler):
1435 extract_map = util.update_copy(
1436 compiler.SQLCompiler.extract_map,
1437 {
1438 "month": "%m",
1439 "day": "%d",
1440 "year": "%Y",
1441 "second": "%S",
1442 "hour": "%H",
1443 "doy": "%j",
1444 "minute": "%M",
1445 "epoch": "%s",
1446 "dow": "%w",
1447 "week": "%W",
1448 },
1449 )
1451 def visit_truediv_binary(self, binary, operator, **kw):
1452 return (
1453 self.process(binary.left, **kw)
1454 + " / "
1455 + "(%s + 0.0)" % self.process(binary.right, **kw)
1456 )
1458 def visit_now_func(self, fn, **kw):
1459 return "CURRENT_TIMESTAMP"
1461 def visit_localtimestamp_func(self, func, **kw):
1462 return "DATETIME(CURRENT_TIMESTAMP, 'localtime')"
1464 def visit_true(self, expr, **kw):
1465 return "1"
1467 def visit_false(self, expr, **kw):
1468 return "0"
1470 def visit_char_length_func(self, fn, **kw):
1471 return "length%s" % self.function_argspec(fn)
1473 def visit_aggregate_strings_func(self, fn, **kw):
1474 return super().visit_aggregate_strings_func(
1475 fn, use_function_name="group_concat", **kw
1476 )
1478 def visit_cast(self, cast, **kwargs):
1479 if self.dialect.supports_cast:
1480 return super().visit_cast(cast, **kwargs)
1481 else:
1482 return self.process(cast.clause, **kwargs)
1484 def visit_extract(self, extract, **kw):
1485 try:
1486 return "CAST(STRFTIME('%s', %s) AS INTEGER)" % (
1487 self.extract_map[extract.field],
1488 self.process(extract.expr, **kw),
1489 )
1490 except KeyError as err:
1491 raise exc.CompileError(
1492 "%s is not a valid extract argument." % extract.field
1493 ) from err
1495 def returning_clause(
1496 self,
1497 stmt,
1498 returning_cols,
1499 *,
1500 populate_result_map,
1501 **kw,
1502 ):
1503 kw["include_table"] = False
1504 return super().returning_clause(
1505 stmt, returning_cols, populate_result_map=populate_result_map, **kw
1506 )
1508 def limit_clause(self, select, **kw):
1509 text = ""
1510 if select._limit_clause is not None:
1511 text += "\n LIMIT " + self.process(select._limit_clause, **kw)
1512 if select._offset_clause is not None:
1513 if select._limit_clause is None:
1514 text += "\n LIMIT " + self.process(sql.literal(-1))
1515 text += " OFFSET " + self.process(select._offset_clause, **kw)
1516 else:
1517 text += " OFFSET " + self.process(sql.literal(0), **kw)
1518 return text
1520 def for_update_clause(self, select, **kw):
1521 # sqlite has no "FOR UPDATE" AFAICT
1522 return ""
1524 def update_from_clause(
1525 self, update_stmt, from_table, extra_froms, from_hints, **kw
1526 ):
1527 kw["asfrom"] = True
1528 return "FROM " + ", ".join(
1529 t._compiler_dispatch(self, fromhints=from_hints, **kw)
1530 for t in extra_froms
1531 )
1533 def visit_is_distinct_from_binary(self, binary, operator, **kw):
1534 return "%s IS NOT %s" % (
1535 self.process(binary.left),
1536 self.process(binary.right),
1537 )
1539 def visit_is_not_distinct_from_binary(self, binary, operator, **kw):
1540 return "%s IS %s" % (
1541 self.process(binary.left),
1542 self.process(binary.right),
1543 )
1545 def visit_json_getitem_op_binary(
1546 self, binary, operator, _cast_applied=False, **kw
1547 ):
1548 if (
1549 not _cast_applied
1550 and binary.type._type_affinity is not sqltypes.JSON
1551 ):
1552 kw["_cast_applied"] = True
1553 return self.process(sql.cast(binary, binary.type), **kw)
1555 if binary.type._type_affinity is sqltypes.JSON:
1556 expr = "JSON_QUOTE(JSON_EXTRACT(%s, %s))"
1557 else:
1558 expr = "JSON_EXTRACT(%s, %s)"
1560 return expr % (
1561 self.process(binary.left, **kw),
1562 self.process(binary.right, **kw),
1563 )
1565 def visit_json_path_getitem_op_binary(
1566 self, binary, operator, _cast_applied=False, **kw
1567 ):
1568 if (
1569 not _cast_applied
1570 and binary.type._type_affinity is not sqltypes.JSON
1571 ):
1572 kw["_cast_applied"] = True
1573 return self.process(sql.cast(binary, binary.type), **kw)
1575 if binary.type._type_affinity is sqltypes.JSON:
1576 expr = "JSON_QUOTE(JSON_EXTRACT(%s, %s))"
1577 else:
1578 expr = "JSON_EXTRACT(%s, %s)"
1580 return expr % (
1581 self.process(binary.left, **kw),
1582 self.process(binary.right, **kw),
1583 )
1585 def visit_empty_set_op_expr(self, type_, expand_op, **kw):
1586 # slightly old SQLite versions don't seem to be able to handle
1587 # the empty set impl
1588 return self.visit_empty_set_expr(type_)
1590 def visit_empty_set_expr(self, element_types, **kw):
1591 return "SELECT %s FROM (SELECT %s) WHERE 1!=1" % (
1592 ", ".join("1" for type_ in element_types or [INTEGER()]),
1593 ", ".join("1" for type_ in element_types or [INTEGER()]),
1594 )
1596 def visit_regexp_match_op_binary(self, binary, operator, **kw):
1597 return self._generate_generic_binary(binary, " REGEXP ", **kw)
1599 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
1600 return self._generate_generic_binary(binary, " NOT REGEXP ", **kw)
1602 def _on_conflict_target(self, clause, **kw):
1603 if clause.inferred_target_elements is not None:
1604 target_text = "(%s)" % ", ".join(
1605 (
1606 self.preparer.quote(c)
1607 if isinstance(c, str)
1608 else self.process(c, include_table=False, use_schema=False)
1609 )
1610 for c in clause.inferred_target_elements
1611 )
1612 if clause.inferred_target_whereclause is not None:
1613 target_text += " WHERE %s" % self.process(
1614 clause.inferred_target_whereclause,
1615 include_table=False,
1616 use_schema=False,
1617 literal_execute=True,
1618 )
1620 else:
1621 target_text = ""
1623 return target_text
1625 def visit_on_conflict_do_nothing(self, on_conflict, **kw):
1626 target_text = self._on_conflict_target(on_conflict, **kw)
1628 if target_text:
1629 return "ON CONFLICT %s DO NOTHING" % target_text
1630 else:
1631 return "ON CONFLICT DO NOTHING"
1633 def visit_on_conflict_do_update(self, on_conflict, **kw):
1634 clause = on_conflict
1636 target_text = self._on_conflict_target(on_conflict, **kw)
1638 action_set_ops = []
1640 set_parameters = dict(clause.update_values_to_set)
1641 # create a list of column assignment clauses as tuples
1643 insert_statement = self.stack[-1]["selectable"]
1644 cols = insert_statement.table.c
1645 for c in cols:
1646 col_key = c.key
1648 if col_key in set_parameters:
1649 value = set_parameters.pop(col_key)
1650 elif c in set_parameters:
1651 value = set_parameters.pop(c)
1652 else:
1653 continue
1655 if (
1656 isinstance(value, elements.BindParameter)
1657 and value.type._isnull
1658 ):
1659 value = value._with_binary_element_type(c.type)
1660 value_text = self.process(value.self_group(), use_schema=False)
1662 key_text = self.preparer.quote(c.name)
1663 action_set_ops.append("%s = %s" % (key_text, value_text))
1665 # check for names that don't match columns
1666 if set_parameters:
1667 util.warn(
1668 "Additional column names not matching "
1669 "any column keys in table '%s': %s"
1670 % (
1671 self.current_executable.table.name,
1672 (", ".join("'%s'" % c for c in set_parameters)),
1673 )
1674 )
1675 for k, v in set_parameters.items():
1676 key_text = (
1677 self.preparer.quote(k)
1678 if isinstance(k, str)
1679 else self.process(k, use_schema=False)
1680 )
1681 value_text = self.process(
1682 coercions.expect(roles.ExpressionElementRole, v),
1683 use_schema=False,
1684 )
1685 action_set_ops.append("%s = %s" % (key_text, value_text))
1687 action_text = ", ".join(action_set_ops)
1688 if clause.update_whereclause is not None:
1689 action_text += " WHERE %s" % self.process(
1690 clause.update_whereclause, include_table=True, use_schema=False
1691 )
1693 return "ON CONFLICT %s DO UPDATE SET %s" % (target_text, action_text)
1695 def visit_bitwise_xor_op_binary(self, binary, operator, **kw):
1696 # sqlite has no xor. Use "a XOR b" = "(a | b) - (a & b)".
1697 kw["eager_grouping"] = True
1698 or_ = self._generate_generic_binary(binary, " | ", **kw)
1699 and_ = self._generate_generic_binary(binary, " & ", **kw)
1700 return f"({or_} - {and_})"
1703class SQLiteDDLCompiler(compiler.DDLCompiler):
1704 def get_column_specification(self, column, **kwargs):
1705 coltype = self.dialect.type_compiler_instance.process(
1706 column.type, type_expression=column
1707 )
1708 colspec = self.preparer.format_column(column) + " " + coltype
1709 default = self.get_column_default_string(column)
1710 if default is not None:
1712 if not re.match(r"""^\s*[\'\"\(]""", default) and re.match(
1713 r".*\W.*", default
1714 ):
1715 colspec += f" DEFAULT ({default})"
1716 else:
1717 colspec += f" DEFAULT {default}"
1719 if not column.nullable:
1720 colspec += " NOT NULL"
1722 on_conflict_clause = column.dialect_options["sqlite"][
1723 "on_conflict_not_null"
1724 ]
1725 if on_conflict_clause is not None:
1726 colspec += " ON CONFLICT " + on_conflict_clause
1728 if column.primary_key:
1729 if (
1730 column.autoincrement is True
1731 and len(column.table.primary_key.columns) != 1
1732 ):
1733 raise exc.CompileError(
1734 "SQLite does not support autoincrement for "
1735 "composite primary keys"
1736 )
1738 if (
1739 column.table.dialect_options["sqlite"]["autoincrement"]
1740 and len(column.table.primary_key.columns) == 1
1741 and issubclass(column.type._type_affinity, sqltypes.Integer)
1742 and not column.foreign_keys
1743 ):
1744 colspec += " PRIMARY KEY"
1746 on_conflict_clause = column.dialect_options["sqlite"][
1747 "on_conflict_primary_key"
1748 ]
1749 if on_conflict_clause is not None:
1750 colspec += " ON CONFLICT " + on_conflict_clause
1752 colspec += " AUTOINCREMENT"
1754 if column.computed is not None:
1755 colspec += " " + self.process(column.computed)
1757 return colspec
1759 def visit_primary_key_constraint(self, constraint, **kw):
1760 # for columns with sqlite_autoincrement=True,
1761 # the PRIMARY KEY constraint can only be inline
1762 # with the column itself.
1763 if len(constraint.columns) == 1:
1764 c = list(constraint)[0]
1765 if (
1766 c.primary_key
1767 and c.table.dialect_options["sqlite"]["autoincrement"]
1768 and issubclass(c.type._type_affinity, sqltypes.Integer)
1769 and not c.foreign_keys
1770 ):
1771 return None
1773 text = super().visit_primary_key_constraint(constraint)
1775 on_conflict_clause = constraint.dialect_options["sqlite"][
1776 "on_conflict"
1777 ]
1778 if on_conflict_clause is None and len(constraint.columns) == 1:
1779 on_conflict_clause = list(constraint)[0].dialect_options["sqlite"][
1780 "on_conflict_primary_key"
1781 ]
1783 if on_conflict_clause is not None:
1784 text += " ON CONFLICT " + on_conflict_clause
1786 return text
1788 def visit_unique_constraint(self, constraint, **kw):
1789 text = super().visit_unique_constraint(constraint)
1791 on_conflict_clause = constraint.dialect_options["sqlite"][
1792 "on_conflict"
1793 ]
1794 if on_conflict_clause is None and len(constraint.columns) == 1:
1795 col1 = list(constraint)[0]
1796 if isinstance(col1, schema.SchemaItem):
1797 on_conflict_clause = list(constraint)[0].dialect_options[
1798 "sqlite"
1799 ]["on_conflict_unique"]
1801 if on_conflict_clause is not None:
1802 text += " ON CONFLICT " + on_conflict_clause
1804 return text
1806 def visit_check_constraint(self, constraint, **kw):
1807 text = super().visit_check_constraint(constraint)
1809 on_conflict_clause = constraint.dialect_options["sqlite"][
1810 "on_conflict"
1811 ]
1813 if on_conflict_clause is not None:
1814 text += " ON CONFLICT " + on_conflict_clause
1816 return text
1818 def visit_column_check_constraint(self, constraint, **kw):
1819 text = super().visit_column_check_constraint(constraint)
1821 if constraint.dialect_options["sqlite"]["on_conflict"] is not None:
1822 raise exc.CompileError(
1823 "SQLite does not support on conflict clause for "
1824 "column check constraint"
1825 )
1827 return text
1829 def visit_foreign_key_constraint(self, constraint, **kw):
1830 local_table = constraint.elements[0].parent.table
1831 remote_table = constraint.elements[0].column.table
1833 if local_table.schema != remote_table.schema:
1834 return None
1835 else:
1836 return super().visit_foreign_key_constraint(constraint)
1838 def define_constraint_remote_table(self, constraint, table, preparer):
1839 """Format the remote table clause of a CREATE CONSTRAINT clause."""
1841 return preparer.format_table(table, use_schema=False)
1843 def visit_create_index(
1844 self, create, include_schema=False, include_table_schema=True, **kw
1845 ):
1846 index = create.element
1847 self._verify_index_table(index)
1848 preparer = self.preparer
1849 text = "CREATE "
1850 if index.unique:
1851 text += "UNIQUE "
1853 text += "INDEX "
1855 if create.if_not_exists:
1856 text += "IF NOT EXISTS "
1858 text += "%s ON %s (%s)" % (
1859 self._prepared_index_name(index, include_schema=True),
1860 preparer.format_table(index.table, use_schema=False),
1861 ", ".join(
1862 self.sql_compiler.process(
1863 expr, include_table=False, literal_binds=True
1864 )
1865 for expr in index.expressions
1866 ),
1867 )
1869 whereclause = index.dialect_options["sqlite"]["where"]
1870 if whereclause is not None:
1871 where_compiled = self.sql_compiler.process(
1872 whereclause, include_table=False, literal_binds=True
1873 )
1874 text += " WHERE " + where_compiled
1876 return text
1878 def post_create_table(self, table):
1879 table_options = []
1881 if not table.dialect_options["sqlite"]["with_rowid"]:
1882 table_options.append("WITHOUT ROWID")
1884 if table.dialect_options["sqlite"]["strict"]:
1885 table_options.append("STRICT")
1887 if table_options:
1888 return "\n " + ",\n ".join(table_options)
1889 else:
1890 return ""
1892 def visit_create_view(self, create, **kw):
1893 """Handle SQLite if_not_exists dialect option for CREATE VIEW."""
1894 # Get the if_not_exists dialect option from the CreateView object
1895 if_not_exists = create.dialect_options["sqlite"].get(
1896 "if_not_exists", False
1897 )
1899 # Pass if_not_exists through kw to the parent's _generate_table_select
1900 kw["if_not_exists"] = if_not_exists
1901 return super().visit_create_view(create, **kw)
1904class SQLiteTypeCompiler(compiler.GenericTypeCompiler):
1905 def visit_large_binary(self, type_, **kw):
1906 return self.visit_BLOB(type_)
1908 def visit_DATETIME(self, type_, **kw):
1909 if (
1910 not isinstance(type_, _DateTimeMixin)
1911 or type_.format_is_text_affinity
1912 ):
1913 return super().visit_DATETIME(type_)
1914 else:
1915 return "DATETIME_CHAR"
1917 def visit_DATE(self, type_, **kw):
1918 if (
1919 not isinstance(type_, _DateTimeMixin)
1920 or type_.format_is_text_affinity
1921 ):
1922 return super().visit_DATE(type_)
1923 else:
1924 return "DATE_CHAR"
1926 def visit_TIME(self, type_, **kw):
1927 if (
1928 not isinstance(type_, _DateTimeMixin)
1929 or type_.format_is_text_affinity
1930 ):
1931 return super().visit_TIME(type_)
1932 else:
1933 return "TIME_CHAR"
1935 def visit_JSON(self, type_, **kw):
1936 # note this name provides NUMERIC affinity, not TEXT.
1937 # should not be an issue unless the JSON value consists of a single
1938 # numeric value. JSONTEXT can be used if this case is required.
1939 return "JSON"
1942class SQLiteIdentifierPreparer(compiler.IdentifierPreparer):
1943 reserved_words = {
1944 "add",
1945 "after",
1946 "all",
1947 "alter",
1948 "analyze",
1949 "and",
1950 "as",
1951 "asc",
1952 "attach",
1953 "autoincrement",
1954 "before",
1955 "begin",
1956 "between",
1957 "by",
1958 "cascade",
1959 "case",
1960 "cast",
1961 "check",
1962 "collate",
1963 "column",
1964 "commit",
1965 "conflict",
1966 "constraint",
1967 "create",
1968 "cross",
1969 "current_date",
1970 "current_time",
1971 "current_timestamp",
1972 "database",
1973 "default",
1974 "deferrable",
1975 "deferred",
1976 "delete",
1977 "desc",
1978 "detach",
1979 "distinct",
1980 "drop",
1981 "each",
1982 "else",
1983 "end",
1984 "escape",
1985 "except",
1986 "exclusive",
1987 "exists",
1988 "explain",
1989 "false",
1990 "fail",
1991 "for",
1992 "foreign",
1993 "from",
1994 "full",
1995 "glob",
1996 "group",
1997 "having",
1998 "if",
1999 "ignore",
2000 "immediate",
2001 "in",
2002 "index",
2003 "indexed",
2004 "initially",
2005 "inner",
2006 "insert",
2007 "instead",
2008 "intersect",
2009 "into",
2010 "is",
2011 "isnull",
2012 "join",
2013 "key",
2014 "left",
2015 "like",
2016 "limit",
2017 "match",
2018 "natural",
2019 "not",
2020 "notnull",
2021 "null",
2022 "of",
2023 "offset",
2024 "on",
2025 "or",
2026 "order",
2027 "outer",
2028 "plan",
2029 "pragma",
2030 "primary",
2031 "query",
2032 "raise",
2033 "references",
2034 "reindex",
2035 "rename",
2036 "replace",
2037 "restrict",
2038 "right",
2039 "rollback",
2040 "row",
2041 "select",
2042 "set",
2043 "table",
2044 "temp",
2045 "temporary",
2046 "then",
2047 "to",
2048 "transaction",
2049 "trigger",
2050 "true",
2051 "union",
2052 "unique",
2053 "update",
2054 "using",
2055 "vacuum",
2056 "values",
2057 "view",
2058 "virtual",
2059 "when",
2060 "where",
2061 }
2064class SQLiteExecutionContext(default.DefaultExecutionContext):
2065 @util.memoized_property
2066 def _preserve_raw_colnames(self):
2067 return (
2068 not self.dialect._broken_dotted_colnames
2069 or self.execution_options.get("sqlite_raw_colnames", False)
2070 )
2072 def _translate_colname(self, colname):
2073 # TODO: detect SQLite version 3.10.0 or greater;
2074 # see [ticket:3633]
2076 # adjust for dotted column names. SQLite
2077 # in the case of UNION may store col names as
2078 # "tablename.colname", or if using an attached database,
2079 # "database.tablename.colname", in cursor.description
2080 if not self._preserve_raw_colnames and "." in colname:
2081 return colname.split(".")[-1], colname
2082 else:
2083 return colname, None
2086class SQLiteDialect(default.DefaultDialect):
2087 name = "sqlite"
2088 supports_alter = False
2090 # SQlite supports "DEFAULT VALUES" but *does not* support
2091 # "VALUES (DEFAULT)"
2092 supports_default_values = True
2093 supports_default_metavalue = False
2095 # sqlite issue:
2096 # https://github.com/python/cpython/issues/93421
2097 # note this parameter is no longer used by the ORM or default dialect
2098 # see #9414
2099 supports_sane_rowcount_returning = False
2101 supports_empty_insert = False
2102 supports_cast = True
2103 supports_multivalues_insert = True
2104 use_insertmanyvalues = True
2105 tuple_in_values = True
2106 supports_statement_cache = True
2107 insert_null_pk_still_autoincrements = True
2108 insert_returning = True
2109 update_returning = True
2110 update_returning_multifrom = True
2111 delete_returning = True
2112 update_returning_multifrom = True
2114 supports_default_metavalue = True
2115 """dialect supports INSERT... VALUES (DEFAULT) syntax"""
2117 default_metavalue_token = "NULL"
2118 """for INSERT... VALUES (DEFAULT) syntax, the token to put in the
2119 parenthesis."""
2121 default_paramstyle = "qmark"
2122 execution_ctx_cls = SQLiteExecutionContext
2123 statement_compiler = SQLiteCompiler
2124 ddl_compiler = SQLiteDDLCompiler
2125 type_compiler_cls = SQLiteTypeCompiler
2126 preparer = SQLiteIdentifierPreparer
2127 ischema_names = ischema_names
2128 colspecs = colspecs
2130 construct_arguments = [
2131 (
2132 sa_schema.Table,
2133 {
2134 "autoincrement": False,
2135 "with_rowid": True,
2136 "strict": False,
2137 },
2138 ),
2139 (sa_schema.Index, {"where": None}),
2140 (
2141 sa_schema.Column,
2142 {
2143 "on_conflict_primary_key": None,
2144 "on_conflict_not_null": None,
2145 "on_conflict_unique": None,
2146 },
2147 ),
2148 (sa_schema.Constraint, {"on_conflict": None}),
2149 (sa_ddl.CreateView, {"if_not_exists": False}),
2150 ]
2152 _broken_fk_pragma_quotes = False
2153 _broken_dotted_colnames = False
2155 def __init__(
2156 self,
2157 native_datetime: bool = False,
2158 json_serializer: Optional[Callable[..., Any]] = None,
2159 json_deserializer: Optional[Callable[..., Any]] = None,
2160 **kwargs: Any,
2161 ) -> None:
2162 default.DefaultDialect.__init__(self, **kwargs)
2164 self._json_serializer = json_serializer
2165 self._json_deserializer = json_deserializer
2167 # this flag used by pysqlite dialect, and perhaps others in the
2168 # future, to indicate the driver is handling date/timestamp
2169 # conversions (and perhaps datetime/time as well on some hypothetical
2170 # driver ?)
2171 self.native_datetime = native_datetime
2173 if self.dbapi is not None:
2174 if self.dbapi.sqlite_version_info < (3, 7, 16):
2175 util.warn(
2176 "SQLite version %s is older than 3.7.16, and will not "
2177 "support right nested joins, as are sometimes used in "
2178 "more complex ORM scenarios. SQLAlchemy 1.4 and above "
2179 "no longer tries to rewrite these joins."
2180 % (self.dbapi.sqlite_version_info,)
2181 )
2183 # NOTE: python 3.7 on fedora for me has SQLite 3.34.1. These
2184 # version checks are getting very stale.
2185 self._broken_dotted_colnames = self.dbapi.sqlite_version_info < (
2186 3,
2187 10,
2188 0,
2189 )
2190 self.supports_default_values = self.dbapi.sqlite_version_info >= (
2191 3,
2192 3,
2193 8,
2194 )
2195 self.supports_cast = self.dbapi.sqlite_version_info >= (3, 2, 3)
2196 self.supports_multivalues_insert = (
2197 # https://www.sqlite.org/releaselog/3_7_11.html
2198 self.dbapi.sqlite_version_info
2199 >= (3, 7, 11)
2200 )
2201 # see https://www.sqlalchemy.org/trac/ticket/2568
2202 # as well as https://www.sqlite.org/src/info/600482d161
2203 self._broken_fk_pragma_quotes = self.dbapi.sqlite_version_info < (
2204 3,
2205 6,
2206 14,
2207 )
2209 if self.dbapi.sqlite_version_info < (3, 35) or util.pypy:
2210 self.update_returning = self.delete_returning = (
2211 self.insert_returning
2212 ) = False
2214 if self.dbapi.sqlite_version_info < (3, 32, 0):
2215 # https://www.sqlite.org/limits.html
2216 self.insertmanyvalues_max_parameters = 999
2218 _isolation_lookup = util.immutabledict(
2219 {"READ UNCOMMITTED": 1, "SERIALIZABLE": 0}
2220 )
2222 def get_isolation_level_values(self, dbapi_connection):
2223 return list(self._isolation_lookup)
2225 def set_isolation_level(
2226 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
2227 ) -> None:
2228 isolation_level = self._isolation_lookup[level]
2230 cursor = dbapi_connection.cursor()
2231 cursor.execute(f"PRAGMA read_uncommitted = {isolation_level}")
2232 cursor.close()
2234 def get_isolation_level(self, dbapi_connection):
2235 cursor = dbapi_connection.cursor()
2236 cursor.execute("PRAGMA read_uncommitted")
2237 res = cursor.fetchone()
2238 if res:
2239 value = res[0]
2240 else:
2241 # https://www.sqlite.org/changes.html#version_3_3_3
2242 # "Optional READ UNCOMMITTED isolation (instead of the
2243 # default isolation level of SERIALIZABLE) and
2244 # table level locking when database connections
2245 # share a common cache.""
2246 # pre-SQLite 3.3.0 default to 0
2247 value = 0
2248 cursor.close()
2249 if value == 0:
2250 return "SERIALIZABLE"
2251 elif value == 1:
2252 return "READ UNCOMMITTED"
2253 else:
2254 assert False, "Unknown isolation level %s" % value
2256 @reflection.cache
2257 def get_schema_names(self, connection, **kw):
2258 s = "PRAGMA database_list"
2259 dl = connection.exec_driver_sql(s)
2261 return [db[1] for db in dl if db[1] != "temp"]
2263 def _format_schema(self, schema, table_name):
2264 if schema is not None:
2265 qschema = self.identifier_preparer.quote_identifier(schema)
2266 name = f"{qschema}.{table_name}"
2267 else:
2268 name = table_name
2269 return name
2271 def _sqlite_main_query(
2272 self,
2273 table: str,
2274 type_: str,
2275 schema: Optional[str],
2276 sqlite_include_internal: bool,
2277 ):
2278 main = self._format_schema(schema, table)
2279 if not sqlite_include_internal:
2280 filter_table = " AND name NOT LIKE 'sqlite~_%' ESCAPE '~'"
2281 else:
2282 filter_table = ""
2283 query = (
2284 f"SELECT name FROM {main} "
2285 f"WHERE type='{type_}'{filter_table} "
2286 "ORDER BY name"
2287 )
2288 return query
2290 @reflection.cache
2291 def get_table_names(
2292 self, connection, schema=None, sqlite_include_internal=False, **kw
2293 ):
2294 query = self._sqlite_main_query(
2295 "sqlite_master", "table", schema, sqlite_include_internal
2296 )
2297 names = connection.exec_driver_sql(query).scalars().all()
2298 return names
2300 @reflection.cache
2301 def get_temp_table_names(
2302 self, connection, sqlite_include_internal=False, **kw
2303 ):
2304 query = self._sqlite_main_query(
2305 "sqlite_temp_master", "table", None, sqlite_include_internal
2306 )
2307 names = connection.exec_driver_sql(query).scalars().all()
2308 return names
2310 @reflection.cache
2311 def get_temp_view_names(
2312 self, connection, sqlite_include_internal=False, **kw
2313 ):
2314 query = self._sqlite_main_query(
2315 "sqlite_temp_master", "view", None, sqlite_include_internal
2316 )
2317 names = connection.exec_driver_sql(query).scalars().all()
2318 return names
2320 @reflection.cache
2321 def has_table(self, connection, table_name, schema=None, **kw):
2322 self._ensure_has_table_connection(connection)
2324 if schema is not None and schema not in self.get_schema_names(
2325 connection, **kw
2326 ):
2327 return False
2329 info = self._get_table_pragma(
2330 connection, "table_info", table_name, schema=schema
2331 )
2332 return bool(info)
2334 def _get_default_schema_name(self, connection):
2335 return "main"
2337 @reflection.cache
2338 def get_view_names(
2339 self, connection, schema=None, sqlite_include_internal=False, **kw
2340 ):
2341 query = self._sqlite_main_query(
2342 "sqlite_master", "view", schema, sqlite_include_internal
2343 )
2344 names = connection.exec_driver_sql(query).scalars().all()
2345 return names
2347 @reflection.cache
2348 def get_view_definition(self, connection, view_name, schema=None, **kw):
2349 if schema is not None:
2350 qschema = self.identifier_preparer.quote_identifier(schema)
2351 master = f"{qschema}.sqlite_master"
2352 s = ("SELECT sql FROM %s WHERE name = ? AND type='view'") % (
2353 master,
2354 )
2355 rs = connection.exec_driver_sql(s, (view_name,))
2356 else:
2357 try:
2358 s = (
2359 "SELECT sql FROM "
2360 " (SELECT * FROM sqlite_master UNION ALL "
2361 " SELECT * FROM sqlite_temp_master) "
2362 "WHERE name = ? "
2363 "AND type='view'"
2364 )
2365 rs = connection.exec_driver_sql(s, (view_name,))
2366 except exc.DBAPIError:
2367 s = (
2368 "SELECT sql FROM sqlite_master WHERE name = ? "
2369 "AND type='view'"
2370 )
2371 rs = connection.exec_driver_sql(s, (view_name,))
2373 result = rs.fetchall()
2374 if result:
2375 return result[0].sql
2376 else:
2377 raise exc.NoSuchTableError(
2378 f"{schema}.{view_name}" if schema else view_name
2379 )
2381 @reflection.cache
2382 def get_columns(self, connection, table_name, schema=None, **kw):
2383 pragma = "table_info"
2384 # computed columns are threaded as hidden, they require table_xinfo
2385 if self.server_version_info >= (3, 31):
2386 pragma = "table_xinfo"
2387 info = self._get_table_pragma(
2388 connection, pragma, table_name, schema=schema
2389 )
2390 columns = []
2391 tablesql = None
2392 for row in info:
2393 name = row[1]
2394 type_ = row[2].upper()
2395 nullable = not row[3]
2396 default = row[4]
2397 primary_key = row[5]
2398 hidden = row[6] if pragma == "table_xinfo" else 0
2400 # hidden has value 0 for normal columns, 1 for hidden columns,
2401 # 2 for computed virtual columns and 3 for computed stored columns
2402 # https://www.sqlite.org/src/info/069351b85f9a706f60d3e98fbc8aaf40c374356b967c0464aede30ead3d9d18b
2403 if hidden == 1:
2404 continue
2406 generated = bool(hidden)
2407 persisted = hidden == 3
2409 if tablesql is None and generated:
2410 tablesql = self._get_table_sql(
2411 connection, table_name, schema, **kw
2412 )
2413 # remove create table
2414 match = re.match(
2415 (
2416 r"create table .*?\((.*)\)"
2417 r"(?:\s*,?\s*(?:WITHOUT\s+ROWID|STRICT))*$"
2418 ),
2419 tablesql.strip(),
2420 re.DOTALL | re.IGNORECASE,
2421 )
2422 assert match, f"create table not found in {tablesql}"
2423 tablesql = match.group(1).strip()
2425 columns.append(
2426 self._get_column_info(
2427 name,
2428 type_,
2429 nullable,
2430 default,
2431 primary_key,
2432 generated,
2433 persisted,
2434 tablesql,
2435 )
2436 )
2437 if columns:
2438 return columns
2439 elif not self.has_table(connection, table_name, schema):
2440 raise exc.NoSuchTableError(
2441 f"{schema}.{table_name}" if schema else table_name
2442 )
2443 else:
2444 return ReflectionDefaults.columns()
2446 def _get_column_info(
2447 self,
2448 name,
2449 type_,
2450 nullable,
2451 default,
2452 primary_key,
2453 generated,
2454 persisted,
2455 tablesql,
2456 ):
2457 if generated:
2458 # the type of a column "cc INTEGER GENERATED ALWAYS AS (1 + 42)"
2459 # somehow is "INTEGER GENERATED ALWAYS"
2460 type_ = re.sub("generated", "", type_, flags=re.IGNORECASE)
2461 type_ = re.sub("always", "", type_, flags=re.IGNORECASE).strip()
2463 coltype = self._resolve_type_affinity(type_)
2465 if default is not None:
2466 default = str(default)
2468 colspec = {
2469 "name": name,
2470 "type": coltype,
2471 "nullable": nullable,
2472 "default": default,
2473 "primary_key": primary_key,
2474 }
2475 if generated:
2476 sqltext = ""
2477 if tablesql:
2478 pattern = (
2479 r"[^,]*\s+GENERATED\s+ALWAYS\s+AS"
2480 r"\s+\((.*)\)\s*(?:virtual|stored)?"
2481 )
2482 match = re.search(
2483 re.escape(name) + pattern, tablesql, re.IGNORECASE
2484 )
2485 if match:
2486 sqltext = match.group(1)
2487 colspec["computed"] = {"sqltext": sqltext, "persisted": persisted}
2488 return colspec
2490 def _resolve_type_affinity(self, type_):
2491 """Return a data type from a reflected column, using affinity rules.
2493 SQLite's goal for universal compatibility introduces some complexity
2494 during reflection, as a column's defined type might not actually be a
2495 type that SQLite understands - or indeed, my not be defined *at all*.
2496 Internally, SQLite handles this with a 'data type affinity' for each
2497 column definition, mapping to one of 'TEXT', 'NUMERIC', 'INTEGER',
2498 'REAL', or 'NONE' (raw bits). The algorithm that determines this is
2499 listed in https://www.sqlite.org/datatype3.html section 2.1.
2501 This method allows SQLAlchemy to support that algorithm, while still
2502 providing access to smarter reflection utilities by recognizing
2503 column definitions that SQLite only supports through affinity (like
2504 DATE and DOUBLE).
2506 """
2507 match = re.match(r"([\w ]+)(\(.*?\))?", type_)
2508 if match:
2509 coltype = match.group(1)
2510 args = match.group(2)
2511 else:
2512 coltype = ""
2513 args = ""
2515 if coltype in self.ischema_names:
2516 coltype = self.ischema_names[coltype]
2517 elif "INT" in coltype:
2518 coltype = sqltypes.INTEGER
2519 elif "CHAR" in coltype or "CLOB" in coltype or "TEXT" in coltype:
2520 coltype = sqltypes.TEXT
2521 elif "BLOB" in coltype or not coltype:
2522 coltype = sqltypes.NullType
2523 elif "REAL" in coltype or "FLOA" in coltype or "DOUB" in coltype:
2524 coltype = sqltypes.REAL
2525 else:
2526 coltype = sqltypes.NUMERIC
2528 if args is not None:
2529 args = re.findall(r"(\d+)", args)
2530 try:
2531 coltype = coltype(*[int(a) for a in args])
2532 except TypeError:
2533 util.warn(
2534 "Could not instantiate type %s with "
2535 "reflected arguments %s; using no arguments."
2536 % (coltype, args)
2537 )
2538 coltype = coltype()
2539 else:
2540 coltype = coltype()
2542 return coltype
2544 @reflection.cache
2545 def get_pk_constraint(self, connection, table_name, schema=None, **kw):
2546 constraint_name = None
2547 table_data = self._get_table_sql(connection, table_name, schema=schema)
2548 if table_data:
2549 PK_PATTERN = r'CONSTRAINT +(?:"(.+?)"|(\w+)) +PRIMARY KEY'
2550 result = re.search(PK_PATTERN, table_data, re.I)
2551 if result:
2552 constraint_name = result.group(1) or result.group(2)
2553 else:
2554 constraint_name = None
2556 cols = self.get_columns(connection, table_name, schema, **kw)
2557 # consider only pk columns. This also avoids sorting the cached
2558 # value returned by get_columns
2559 cols = [col for col in cols if col.get("primary_key", 0) > 0]
2560 cols.sort(key=lambda col: col.get("primary_key"))
2561 pkeys = [col["name"] for col in cols]
2563 if pkeys:
2564 return {"constrained_columns": pkeys, "name": constraint_name}
2565 else:
2566 return ReflectionDefaults.pk_constraint()
2568 @reflection.cache
2569 def get_foreign_keys(self, connection, table_name, schema=None, **kw):
2570 # sqlite makes this *extremely difficult*.
2571 # First, use the pragma to get the actual FKs.
2572 pragma_fks = self._get_table_pragma(
2573 connection, "foreign_key_list", table_name, schema=schema
2574 )
2576 fks = {}
2578 for row in pragma_fks:
2579 (numerical_id, rtbl, lcol, rcol) = (row[0], row[2], row[3], row[4])
2581 if not rcol:
2582 # no referred column, which means it was not named in the
2583 # original DDL. The referred columns of the foreign key
2584 # constraint are therefore the primary key of the referred
2585 # table.
2586 try:
2587 referred_pk = self.get_pk_constraint(
2588 connection, rtbl, schema=schema, **kw
2589 )
2590 referred_columns = referred_pk["constrained_columns"]
2591 except exc.NoSuchTableError:
2592 # ignore not existing parents
2593 referred_columns = []
2594 else:
2595 # note we use this list only if this is the first column
2596 # in the constraint. for subsequent columns we ignore the
2597 # list and append "rcol" if present.
2598 referred_columns = []
2600 if self._broken_fk_pragma_quotes:
2601 rtbl = re.sub(r"^[\"\[`\']|[\"\]`\']$", "", rtbl)
2603 if numerical_id in fks:
2604 fk = fks[numerical_id]
2605 else:
2606 fk = fks[numerical_id] = {
2607 "name": None,
2608 "constrained_columns": [],
2609 "referred_schema": schema,
2610 "referred_table": rtbl,
2611 "referred_columns": referred_columns,
2612 "options": {},
2613 }
2614 fks[numerical_id] = fk
2616 fk["constrained_columns"].append(lcol)
2618 if rcol:
2619 fk["referred_columns"].append(rcol)
2621 def fk_sig(constrained_columns, referred_table, referred_columns):
2622 return (
2623 tuple(constrained_columns)
2624 + (referred_table,)
2625 + tuple(referred_columns)
2626 )
2628 # then, parse the actual SQL and attempt to find DDL that matches
2629 # the names as well. SQLite saves the DDL in whatever format
2630 # it was typed in as, so need to be liberal here.
2632 keys_by_signature = {
2633 fk_sig(
2634 fk["constrained_columns"],
2635 fk["referred_table"],
2636 fk["referred_columns"],
2637 ): fk
2638 for fk in fks.values()
2639 }
2641 table_data = self._get_table_sql(connection, table_name, schema=schema)
2643 def parse_fks():
2644 if table_data is None:
2645 # system tables, etc.
2646 return
2648 # note that we already have the FKs from PRAGMA above. This whole
2649 # regexp thing is trying to locate additional detail about the
2650 # FKs, namely the name of the constraint and other options.
2651 # so parsing the columns is really about matching it up to what
2652 # we already have.
2653 FK_PATTERN = (
2654 r'(?:CONSTRAINT +(?:"(.+?)"|(\w+)) +)?'
2655 r"FOREIGN KEY *\( *(.+?) *\) +"
2656 r'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\( *((?:(?:"[^"]+"|[a-z0-9_]+) *(?:, *)?)+)\) *' # noqa: E501
2657 r"((?:ON (?:DELETE|UPDATE) "
2658 r"(?:SET NULL|SET DEFAULT|CASCADE|RESTRICT|NO ACTION) *)*)"
2659 r"((?:NOT +)?DEFERRABLE)?"
2660 r"(?: +INITIALLY +(DEFERRED|IMMEDIATE))?"
2661 )
2662 for match in re.finditer(FK_PATTERN, table_data, re.I):
2663 (
2664 constraint_quoted_name,
2665 constraint_name,
2666 constrained_columns,
2667 referred_quoted_name,
2668 referred_name,
2669 referred_columns,
2670 onupdatedelete,
2671 deferrable,
2672 initially,
2673 ) = match.group(1, 2, 3, 4, 5, 6, 7, 8, 9)
2674 constraint_name = constraint_quoted_name or constraint_name
2675 constrained_columns = list(
2676 self._find_cols_in_sig(constrained_columns)
2677 )
2678 if not referred_columns:
2679 referred_columns = constrained_columns
2680 else:
2681 referred_columns = list(
2682 self._find_cols_in_sig(referred_columns)
2683 )
2684 referred_name = referred_quoted_name or referred_name
2685 options = {}
2687 for token in re.split(r" *\bON\b *", onupdatedelete.upper()):
2688 if token.startswith("DELETE"):
2689 ondelete = token[6:].strip()
2690 if ondelete and ondelete != "NO ACTION":
2691 options["ondelete"] = ondelete
2692 elif token.startswith("UPDATE"):
2693 onupdate = token[6:].strip()
2694 if onupdate and onupdate != "NO ACTION":
2695 options["onupdate"] = onupdate
2697 if deferrable:
2698 options["deferrable"] = "NOT" not in deferrable.upper()
2699 if initially:
2700 options["initially"] = initially.upper()
2702 yield (
2703 constraint_name,
2704 constrained_columns,
2705 referred_name,
2706 referred_columns,
2707 options,
2708 )
2710 fkeys = []
2712 for (
2713 constraint_name,
2714 constrained_columns,
2715 referred_name,
2716 referred_columns,
2717 options,
2718 ) in parse_fks():
2719 sig = fk_sig(constrained_columns, referred_name, referred_columns)
2720 if sig not in keys_by_signature:
2721 util.warn(
2722 "WARNING: SQL-parsed foreign key constraint "
2723 "'%s' could not be located in PRAGMA "
2724 "foreign_keys for table %s" % (sig, table_name)
2725 )
2726 continue
2727 key = keys_by_signature.pop(sig)
2728 key["name"] = constraint_name
2729 key["options"] = options
2730 fkeys.append(key)
2731 # assume the remainders are the unnamed, inline constraints, just
2732 # use them as is as it's extremely difficult to parse inline
2733 # constraints
2734 fkeys.extend(keys_by_signature.values())
2735 if fkeys:
2736 return fkeys
2737 else:
2738 return ReflectionDefaults.foreign_keys()
2740 def _find_cols_in_sig(self, sig):
2741 for match in re.finditer(r'(?:"(.+?)")|([a-z0-9_]+)', sig, re.I):
2742 yield match.group(1) or match.group(2)
2744 @reflection.cache
2745 def get_unique_constraints(
2746 self, connection, table_name, schema=None, **kw
2747 ):
2748 auto_index_by_sig = {}
2749 for idx in self.get_indexes(
2750 connection,
2751 table_name,
2752 schema=schema,
2753 include_auto_indexes=True,
2754 **kw,
2755 ):
2756 if not idx["name"].startswith("sqlite_autoindex"):
2757 continue
2758 sig = tuple(idx["column_names"])
2759 auto_index_by_sig[sig] = idx
2761 table_data = self._get_table_sql(
2762 connection, table_name, schema=schema, **kw
2763 )
2764 unique_constraints = []
2766 def parse_uqs():
2767 if table_data is None:
2768 return
2769 UNIQUE_PATTERN = (
2770 r'(?:CONSTRAINT +(?:"(.+?)"|(\w+)) +)?UNIQUE *\((.+?)\)'
2771 )
2772 INLINE_UNIQUE_PATTERN = (
2773 r'(?:(".+?")|(?:[\[`])?([a-z0-9_]+)(?:[\]`])?)[\t ]'
2774 r"+[a-z0-9_ ]+?[\t ]+UNIQUE"
2775 )
2777 for match in re.finditer(UNIQUE_PATTERN, table_data, re.I):
2778 quoted_name, unquoted_name, cols = match.group(1, 2, 3)
2779 name = quoted_name or unquoted_name
2780 yield name, list(self._find_cols_in_sig(cols))
2782 # we need to match inlines as well, as we seek to differentiate
2783 # a UNIQUE constraint from a UNIQUE INDEX, even though these
2784 # are kind of the same thing :)
2785 for match in re.finditer(INLINE_UNIQUE_PATTERN, table_data, re.I):
2786 cols = list(
2787 self._find_cols_in_sig(match.group(1) or match.group(2))
2788 )
2789 yield None, cols
2791 for name, cols in parse_uqs():
2792 sig = tuple(cols)
2793 if sig in auto_index_by_sig:
2794 auto_index_by_sig.pop(sig)
2795 parsed_constraint = {"name": name, "column_names": cols}
2796 unique_constraints.append(parsed_constraint)
2797 # NOTE: auto_index_by_sig might not be empty here,
2798 # the PRIMARY KEY may have an entry.
2799 if unique_constraints:
2800 return unique_constraints
2801 else:
2802 return ReflectionDefaults.unique_constraints()
2804 @reflection.cache
2805 def get_check_constraints(self, connection, table_name, schema=None, **kw):
2806 table_data = self._get_table_sql(
2807 connection, table_name, schema=schema, **kw
2808 )
2810 # Extract CHECK constraints by properly handling balanced parentheses
2811 # and avoiding false matches when CHECK/CONSTRAINT appear in table
2812 # names. See #12924 for context.
2813 #
2814 # SQLite supports 4 identifier quote styles (see
2815 # sqlite.org/lang_keywords.html):
2816 # - Double quotes "..." (standard SQL)
2817 # - Brackets [...] (MS Access/SQL Server compatibility)
2818 # - Backticks `...` (MySQL compatibility)
2819 # - Single quotes '...' (SQLite extension)
2820 #
2821 # NOTE: there is not currently a way to parse CHECK constraints that
2822 # contain newlines as the approach here relies upon each individual
2823 # CHECK constraint being on a single line by itself. This necessarily
2824 # makes assumptions as to how the CREATE TABLE was emitted.
2825 CHECK_PATTERN = re.compile(
2826 r"""
2827 (?<![A-Za-z0-9_]) # Negative lookbehind: ensure CHECK is not
2828 # part of an identifier (e.g., table name
2829 # like "tableCHECK")
2831 (?: # Optional CONSTRAINT clause
2832 CONSTRAINT\s+
2833 ( # Group 1: Constraint name (quoted or unquoted)
2834 "(?:[^"]|"")+" # Double-quoted: "name" or "na""me"
2835 |'(?:[^']|'')+' # Single-quoted: 'name' or 'na''me'
2836 |\[(?:[^\]]|\]\])+\] # Bracket-quoted: [name] or [na]]me]
2837 |`(?:[^`]|``)+` # Backtick-quoted: `name` or `na``me`
2838 |\S+ # Unquoted: simple_name
2839 )
2840 \s+
2841 )?
2843 CHECK\s*\( # CHECK keyword followed by opening paren
2844 """,
2845 re.VERBOSE | re.IGNORECASE,
2846 )
2847 cks = []
2849 for match in re.finditer(CHECK_PATTERN, table_data or ""):
2850 constraint_name = match.group(1)
2852 if constraint_name:
2853 # Remove surrounding quotes if present
2854 # Double quotes: "name" -> name
2855 # Single quotes: 'name' -> name
2856 # Brackets: [name] -> name
2857 # Backticks: `name` -> name
2858 constraint_name = re.sub(
2859 r'^(["\'`])(.+)\1$|^\[(.+)\]$',
2860 lambda m: m.group(2) or m.group(3),
2861 constraint_name,
2862 flags=re.DOTALL,
2863 )
2865 # Find the matching closing parenthesis by counting balanced parens
2866 # Must track string context to ignore parens inside string literals
2867 start = match.end() # Position after 'CHECK ('
2868 paren_count = 1
2869 in_single_quote = False
2870 in_double_quote = False
2872 for pos, char in enumerate(table_data[start:], start):
2873 # Track string literal context
2874 if char == "'" and not in_double_quote:
2875 in_single_quote = not in_single_quote
2876 elif char == '"' and not in_single_quote:
2877 in_double_quote = not in_double_quote
2878 # Only count parens when not inside a string literal
2879 elif not in_single_quote and not in_double_quote:
2880 if char == "(":
2881 paren_count += 1
2882 elif char == ")":
2883 paren_count -= 1
2884 if paren_count == 0:
2885 # Successfully found matching closing parenthesis
2886 sqltext = table_data[start:pos].strip()
2887 cks.append(
2888 {"sqltext": sqltext, "name": constraint_name}
2889 )
2890 break
2892 cks.sort(key=lambda d: d["name"] or "~") # sort None as last
2893 if cks:
2894 return cks
2895 else:
2896 return ReflectionDefaults.check_constraints()
2898 @reflection.cache
2899 def get_indexes(self, connection, table_name, schema=None, **kw):
2900 pragma_indexes = self._get_table_pragma(
2901 connection, "index_list", table_name, schema=schema
2902 )
2903 indexes = []
2905 # regular expression to extract the filter predicate of a partial
2906 # index. this could fail to extract the predicate correctly on
2907 # indexes created like
2908 # CREATE INDEX i ON t (col || ') where') WHERE col <> ''
2909 # but as this function does not support expression-based indexes
2910 # this case does not occur.
2911 partial_pred_re = re.compile(r"\)\s+where\s+(.+)", re.IGNORECASE)
2913 if schema:
2914 schema_expr = "%s." % self.identifier_preparer.quote_identifier(
2915 schema
2916 )
2917 else:
2918 schema_expr = ""
2920 include_auto_indexes = kw.pop("include_auto_indexes", False)
2921 for row in pragma_indexes:
2922 # ignore implicit primary key index.
2923 # https://www.mail-archive.com/sqlite-users@sqlite.org/msg30517.html
2924 if not include_auto_indexes and row[1].startswith(
2925 "sqlite_autoindex"
2926 ):
2927 continue
2928 indexes.append(
2929 dict(
2930 name=row[1],
2931 column_names=[],
2932 unique=row[2],
2933 dialect_options={},
2934 )
2935 )
2937 # check partial indexes
2938 if len(row) >= 5 and row[4]:
2939 s = (
2940 "SELECT sql FROM %(schema)ssqlite_master "
2941 "WHERE name = ? "
2942 "AND type = 'index'" % {"schema": schema_expr}
2943 )
2944 rs = connection.exec_driver_sql(s, (row[1],))
2945 index_sql = rs.scalar()
2946 predicate_match = partial_pred_re.search(index_sql)
2947 if predicate_match is None:
2948 # unless the regex is broken this case shouldn't happen
2949 # because we know this is a partial index, so the
2950 # definition sql should match the regex
2951 util.warn(
2952 "Failed to look up filter predicate of "
2953 "partial index %s" % row[1]
2954 )
2955 else:
2956 predicate = predicate_match.group(1)
2957 indexes[-1]["dialect_options"]["sqlite_where"] = text(
2958 predicate
2959 )
2961 # loop thru unique indexes to get the column names.
2962 for idx in list(indexes):
2963 pragma_index = self._get_table_pragma(
2964 connection, "index_info", idx["name"], schema=schema
2965 )
2967 for row in pragma_index:
2968 if row[2] is None:
2969 util.warn(
2970 "Skipped unsupported reflection of "
2971 "expression-based index %s" % idx["name"]
2972 )
2973 indexes.remove(idx)
2974 break
2975 else:
2976 idx["column_names"].append(row[2])
2978 indexes.sort(key=lambda d: d["name"] or "~") # sort None as last
2979 if indexes:
2980 return indexes
2981 elif not self.has_table(connection, table_name, schema):
2982 raise exc.NoSuchTableError(
2983 f"{schema}.{table_name}" if schema else table_name
2984 )
2985 else:
2986 return ReflectionDefaults.indexes()
2988 def _is_sys_table(self, table_name):
2989 return table_name in {
2990 "sqlite_schema",
2991 "sqlite_master",
2992 "sqlite_temp_schema",
2993 "sqlite_temp_master",
2994 }
2996 @reflection.cache
2997 def _get_table_sql(self, connection, table_name, schema=None, **kw):
2998 if schema:
2999 schema_expr = "%s." % (
3000 self.identifier_preparer.quote_identifier(schema)
3001 )
3002 else:
3003 schema_expr = ""
3004 try:
3005 s = (
3006 "SELECT sql FROM "
3007 " (SELECT * FROM %(schema)ssqlite_master UNION ALL "
3008 " SELECT * FROM %(schema)ssqlite_temp_master) "
3009 "WHERE name = ? "
3010 "AND type in ('table', 'view')" % {"schema": schema_expr}
3011 )
3012 rs = connection.exec_driver_sql(s, (table_name,))
3013 except exc.DBAPIError:
3014 s = (
3015 "SELECT sql FROM %(schema)ssqlite_master "
3016 "WHERE name = ? "
3017 "AND type in ('table', 'view')" % {"schema": schema_expr}
3018 )
3019 rs = connection.exec_driver_sql(s, (table_name,))
3020 value = rs.scalar()
3021 if value is None and not self._is_sys_table(table_name):
3022 raise exc.NoSuchTableError(f"{schema_expr}{table_name}")
3023 return value
3025 def _get_table_pragma(self, connection, pragma, table_name, schema=None):
3026 quote = self.identifier_preparer.quote_identifier
3027 if schema is not None:
3028 statements = [f"PRAGMA {quote(schema)}."]
3029 else:
3030 # because PRAGMA looks in all attached databases if no schema
3031 # given, need to specify "main" schema, however since we want
3032 # 'temp' tables in the same namespace as 'main', need to run
3033 # the PRAGMA twice
3034 statements = ["PRAGMA main.", "PRAGMA temp."]
3036 qtable = quote(table_name)
3037 for statement in statements:
3038 statement = f"{statement}{pragma}({qtable})"
3039 cursor = connection.exec_driver_sql(statement)
3040 if not cursor._soft_closed:
3041 # work around SQLite issue whereby cursor.description
3042 # is blank when PRAGMA returns no rows:
3043 # https://www.sqlite.org/cvstrac/tktview?tn=1884
3044 result = cursor.fetchall()
3045 else:
3046 result = []
3047 if result:
3048 return result
3049 else:
3050 return []