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/pysqlite.py
2# Copyright (C) 2005-2026 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
9r"""
10.. dialect:: sqlite+pysqlite
11 :name: pysqlite
12 :dbapi: sqlite3
13 :connectstring: sqlite+pysqlite:///file_path
14 :url: https://docs.python.org/library/sqlite3.html
16 Note that ``pysqlite`` is the same driver as the ``sqlite3``
17 module included with the Python distribution.
19Driver
20------
22The ``sqlite3`` Python DBAPI is standard on all modern Python versions;
23for cPython and Pypy, no additional installation is necessary.
26Connect Strings
27---------------
29The file specification for the SQLite database is taken as the "database"
30portion of the URL. Note that the format of a SQLAlchemy url is:
32.. sourcecode:: text
34 driver://user:pass@host/database
36This means that the actual filename to be used starts with the characters to
37the **right** of the third slash. So connecting to a relative filepath
38looks like::
40 # relative path
41 e = create_engine("sqlite:///path/to/database.db")
43An absolute path, which is denoted by starting with a slash, means you
44need **four** slashes::
46 # absolute path
47 e = create_engine("sqlite:////path/to/database.db")
49To use a Windows path, regular drive specifications and backslashes can be
50used. Double backslashes are probably needed::
52 # absolute path on Windows
53 e = create_engine("sqlite:///C:\\path\\to\\database.db")
55To use sqlite ``:memory:`` database specify it as the filename using
56``sqlite:///:memory:``. It's also the default if no filepath is
57present, specifying only ``sqlite://`` and nothing else::
59 # in-memory database (note three slashes)
60 e = create_engine("sqlite:///:memory:")
61 # also in-memory database
62 e2 = create_engine("sqlite://")
64.. _pysqlite_uri_connections:
66URI Connections
67^^^^^^^^^^^^^^^
69Modern versions of SQLite support an alternative system of connecting using a
70`driver level URI <https://www.sqlite.org/uri.html>`_, which has the advantage
71that additional driver-level arguments can be passed including options such as
72"read only". The Python sqlite3 driver supports this mode under modern Python
733 versions. The SQLAlchemy pysqlite driver supports this mode of use by
74specifying "uri=true" in the URL query string. The SQLite-level "URI" is kept
75as the "database" portion of the SQLAlchemy url (that is, following a slash)::
77 e = create_engine("sqlite:///file:path/to/database?mode=ro&uri=true")
79.. note:: The "uri=true" parameter must appear in the **query string**
80 of the URL. It will not currently work as expected if it is only
81 present in the :paramref:`_sa.create_engine.connect_args`
82 parameter dictionary.
84The logic reconciles the simultaneous presence of SQLAlchemy's query string and
85SQLite's query string by separating out the parameters that belong to the
86Python sqlite3 driver vs. those that belong to the SQLite URI. This is
87achieved through the use of a fixed list of parameters known to be accepted by
88the Python side of the driver. For example, to include a URL that indicates
89the Python sqlite3 "timeout" and "check_same_thread" parameters, along with the
90SQLite "mode" and "nolock" parameters, they can all be passed together on the
91query string::
93 e = create_engine(
94 "sqlite:///file:path/to/database?"
95 "check_same_thread=true&timeout=10&mode=ro&nolock=1&uri=true"
96 )
98Above, the pysqlite / sqlite3 DBAPI would be passed arguments as::
100 sqlite3.connect(
101 "file:path/to/database?mode=ro&nolock=1",
102 check_same_thread=True,
103 timeout=10,
104 uri=True,
105 )
107Regarding future parameters added to either the Python or native drivers. new
108parameter names added to the SQLite URI scheme should be automatically
109accommodated by this scheme. New parameter names added to the Python driver
110side can be accommodated by specifying them in the
111:paramref:`_sa.create_engine.connect_args` dictionary,
112until dialect support is
113added by SQLAlchemy. For the less likely case that the native SQLite driver
114adds a new parameter name that overlaps with one of the existing, known Python
115driver parameters (such as "timeout" perhaps), SQLAlchemy's dialect would
116require adjustment for the URL scheme to continue to support this.
118As is always the case for all SQLAlchemy dialects, the entire "URL" process
119can be bypassed in :func:`_sa.create_engine` through the use of the
120:paramref:`_sa.create_engine.creator`
121parameter which allows for a custom callable
122that creates a Python sqlite3 driver level connection directly.
124.. seealso::
126 `Uniform Resource Identifiers <https://www.sqlite.org/uri.html>`_ - in
127 the SQLite documentation
129.. _pysqlite_regexp:
131Regular Expression Support
132---------------------------
134.. versionadded:: 1.4
136Support for the :meth:`_sql.ColumnOperators.regexp_match` operator is provided
137using Python's re.search_ function. SQLite itself does not include a working
138regular expression operator; instead, it includes a non-implemented placeholder
139operator ``REGEXP`` that calls a user-defined function that must be provided.
141SQLAlchemy's implementation makes use of the pysqlite create_function_ hook
142as follows::
145 def regexp(a, b):
146 return re.search(a, b) is not None
149 sqlite_connection.create_function(
150 "regexp",
151 2,
152 regexp,
153 )
155There is currently no support for regular expression flags as a separate
156argument, as these are not supported by SQLite's REGEXP operator, however these
157may be included inline within the regular expression string. See `Python regular expressions`_ for
158details.
160.. seealso::
162 `Python regular expressions`_: Documentation for Python's regular expression syntax.
164.. _create_function: https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
166.. _re.search: https://docs.python.org/3/library/re.html#re.search
168.. _Python regular expressions: https://docs.python.org/3/library/re.html#re.search
172Compatibility with sqlite3 "native" date and datetime types
173-----------------------------------------------------------
175The pysqlite driver includes the sqlite3.PARSE_DECLTYPES and
176sqlite3.PARSE_COLNAMES options, which have the effect of any column
177or expression explicitly cast as "date" or "timestamp" will be converted
178to a Python date or datetime object. The date and datetime types provided
179with the pysqlite dialect are not currently compatible with these options,
180since they render the ISO date/datetime including microseconds, which
181pysqlite's driver does not. Additionally, SQLAlchemy does not at
182this time automatically render the "cast" syntax required for the
183freestanding functions "current_timestamp" and "current_date" to return
184datetime/date types natively. Unfortunately, pysqlite
185does not provide the standard DBAPI types in ``cursor.description``,
186leaving SQLAlchemy with no way to detect these types on the fly
187without expensive per-row type checks.
189Keeping in mind that pysqlite's parsing option is not recommended,
190nor should be necessary, for use with SQLAlchemy, usage of PARSE_DECLTYPES
191can be forced if one configures "native_datetime=True" on create_engine()::
193 engine = create_engine(
194 "sqlite://",
195 connect_args={
196 "detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES
197 },
198 native_datetime=True,
199 )
201With this flag enabled, the DATE and TIMESTAMP types (but note - not the
202DATETIME or TIME types...confused yet ?) will not perform any bind parameter
203or result processing. Execution of "func.current_date()" will return a string.
204"func.current_timestamp()" is registered as returning a DATETIME type in
205SQLAlchemy, so this function still receives SQLAlchemy-level result
206processing.
208.. _pysqlite_threading_pooling:
210Concurrency/Threading/Pooling Behavior
211--------------------------------------
213The ``sqlite3`` DBAPI by default prohibits the use of a particular connection
214in a thread which is not the one in which it was created. As SQLite has
215matured, it's behavior under multiple threads has improved, and even includes
216options for memory only databases to be used in multiple threads.
218The thread prohibition is known as "check same thread" and may be controlled
219using the ``sqlite3`` parameter ``check_same_thread``, which will disable or
220enable this check. SQLAlchemy's default behavior here is to set
221``check_same_thread`` to ``False`` automatically whenever a file-based database
222is in use, to establish compatibility with the default pool class
223:class:`.QueuePool`.
225The SQLAlchemy ``pysqlite`` DBAPI establishes the connection pool differently
226based on the kind of SQLite database that's requested:
228* When a ``:memory:`` SQLite database is specified, the dialect by default
229 will use :class:`.SingletonThreadPool`. This pool maintains a single
230 connection per thread, so that all access to the engine within the current
231 thread use the same ``:memory:`` database - other threads would access a
232 different ``:memory:`` database. The ``check_same_thread`` parameter
233 defaults to ``True``.
234* When a file-based database is specified, the dialect will use
235 :class:`.QueuePool` as the source of connections. at the same time,
236 the ``check_same_thread`` flag is set to False by default unless overridden.
238 .. versionchanged:: 2.0
240 SQLite file database engines now use :class:`.QueuePool` by default.
241 Previously, :class:`.NullPool` were used. The :class:`.NullPool` class
242 may be used by specifying it via the
243 :paramref:`_sa.create_engine.poolclass` parameter.
245This selection is made based on the database name alone. Where a
246particular pool class is desired, it should be stated explicitly using the
247:paramref:`_sa.create_engine.poolclass` parameter, in which case no
248selection takes place at all.
250.. deprecated:: 2.1
252 A URL that passes ``mode=memory`` in the query string is currently also
253 given a single-connection pool class. This behavior is deprecated and
254 will be removed in a future release, at which point such URLs will
255 receive :class:`.QueuePool` like any other. This affects URLs such as
256 ``sqlite:///file:mydb?mode=memory&cache=shared&uri=true``, for which
257 :class:`.QueuePool` is in fact the appropriate class, as a shared cache
258 database supports multiple concurrent connections; see
259 :ref:`pysqlite_uri_shared_cache`. Applications relying on the present
260 behavior should state the pool class explicitly.
262Disabling Connection Pooling for File Databases
263^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
265Pooling may be disabled for a file based database by specifying the
266:class:`.NullPool` implementation for the :func:`_sa.create_engine.poolclass`
267parameter::
269 from sqlalchemy import NullPool
271 engine = create_engine("sqlite:///myfile.db", poolclass=NullPool)
273It's been observed that the :class:`.NullPool` implementation incurs an
274extremely small performance overhead for repeated checkouts due to the lack of
275connection reuse implemented by :class:`.QueuePool`. However, it still
276may be beneficial to use this class if the application is experiencing
277issues with files being locked.
279Using a Memory Database in Multiple Threads or Coroutines
280^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
282A ``:memory:`` SQLite database exists only within the scope of a single
283DBAPI connection. It is not possible for two separate ``sqlite3``
284connection objects to access the same ``:memory:`` database unless SQLite's
285shared-cache feature is enabled. Without shared cache, each connection
286creates its own independent in-memory database.
288This means a ``:memory:`` database is **not suitable** for use with
289multiple concurrent threads or coroutines unless either:
291* All workers are fully serialized (mutexed) against each other
292 such that only one is using the database at a time, or
293* SQLite's shared-cache URI feature is used to allow multiple
294 independent connections to access the same in-memory database.
296The same considerations apply when using the :ref:`aiosqlite <aiosqlite>`
297dialect, which wraps ``pysqlite`` connections in an async interface —
298without shared cache, each connection still creates its own independent
299in-memory database.
301The recommended approach for multithreaded or async in-memory
302use is the shared-cache URI feature, described below.
304.. _pysqlite_uri_shared_cache:
306Using a Shared-Cache Memory Database
307~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
309SQLite's
310`URI shared-cache <https://www.sqlite.org/sharedcache.html>`_
311mode allows multiple independent DBAPI connections, each with
312their own transaction state, to access the same in-memory
313database. This is enabled by using a ``file:`` URI with
314``cache=shared`` and passing ``uri=true`` in the query
315string::
317 engine = create_engine("sqlite:///file::memory:?cache=shared&uri=true")
319For async use with aiosqlite, use
320:func:`_asyncio.create_async_engine`::
322 engine = create_async_engine(
323 "sqlite+aiosqlite:///file::memory:?cache=shared&uri=true"
324 )
326Because this URL form is treated as a file-based database by the
327dialect, :class:`.QueuePool` is used automatically and
328``check_same_thread`` defaults to ``False``, so no additional pool
329or connect_args configuration is needed. Each checkout from the
330pool is a distinct DBAPI connection with its own transaction state.
332The shared-cache database is scoped by the filename component of
333the URI. ``file::memory:`` (empty name) is process-global — all
334engines in the process that use this URI share the same database.
335To maintain multiple independent in-memory databases within the
336same process, supply a distinct name for each. A named database of
337this kind requires ``mode=memory``, which presently causes a
338single-connection pool to be selected; :class:`.QueuePool` should
339therefore be requested explicitly::
341 from sqlalchemy.pool import QueuePool
343 engine_a = create_engine(
344 "sqlite:///file:db_a?mode=memory&cache=shared&uri=true",
345 poolclass=QueuePool,
346 )
347 engine_b = create_engine(
348 "sqlite:///file:db_b?mode=memory&cache=shared&uri=true",
349 poolclass=QueuePool,
350 )
352For :func:`_asyncio.create_async_engine`, use
353:class:`.AsyncAdaptedQueuePool` in the same way.
355.. deprecated:: 2.1
357 Selection of a single-connection pool class based on ``mode=memory``
358 is deprecated; a future release will use :class:`.QueuePool` for these
359 URLs, at which point stating
360 :paramref:`_sa.create_engine.poolclass` explicitly will no longer be
361 necessary. See :ref:`pysqlite_threading_pooling`.
363.. _pysqlite_shared_cache_lifespan:
365Lifespan of a Shared-Cache Memory Database
366~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
368A shared-cache in-memory database exists only for as long as at least one
369connection to it remains open; when the last connection is closed, the
370database and all of its contents are discarded. A subsequent connection
371using the same URI then opens a new, empty database, which typically
372surfaces as ``no such table`` errors.
374:class:`.QueuePool` retains connections that have been returned to it, so
375in a default configuration the database will normally persist once the
376first connection has been established. This is a consequence of pool
377behavior rather than a guarantee, however, and the database will be
378discarded by ordinary pool operations including:
380* :meth:`_engine.Engine.dispose`, which closes all connections currently
381 in the pool
382* :paramref:`_sa.create_engine.pool_recycle`, as a recycled connection is
383 closed before its replacement is opened
384* connection invalidation, including that performed by
385 :paramref:`_sa.create_engine.pool_pre_ping`
386* use of :class:`.NullPool`, which closes each connection as it is
387 returned
389Where the database must survive independently of pool activity, hold a
390single connection open for as long as the database is needed::
392 engine = create_engine("sqlite:///file::memory:?cache=shared&uri=true")
394 # keep the database alive for the lifetime of the engine
395 keepalive = engine.connect()
397The same consideration applies to the :ref:`aiosqlite <aiosqlite>`
398dialect, using :meth:`_asyncio.AsyncEngine.connect`.
400Using StaticPool for Single-Connection Memory Databases
401~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
403An older approach for sharing a ``:memory:`` database among threads
404is to force all checkouts to return the same DBAPI connection using
405:class:`.StaticPool`. **This approach does not support any form of
406concurrency** and is only useful when all access to the engine is
407fully serialized, such as in single-threaded test suites::
409 from sqlalchemy.pool import StaticPool
411 engine = create_engine(
412 "sqlite://",
413 connect_args={"check_same_thread": False},
414 poolclass=StaticPool,
415 )
417.. warning::
419 Because :class:`.StaticPool` maintains a single DBAPI connection,
420 all :class:`.Session` or :class:`.Connection` objects that use
421 this engine share that one underlying connection and its single
422 SQLite transaction state. A ``ROLLBACK`` issued by one session
423 (e.g. during error handling) will also roll back uncommitted work
424 from any other session, and concurrent ``COMMIT`` / ``ROLLBACK``
425 calls can interfere with each other unpredictably. This approach
426 is only appropriate when access to the engine is fully serialized,
427 such as in single-threaded test suites. For concurrent workloads,
428 use the shared-cache URI approach described above.
430Using Temporary Tables with SQLite
431^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
433Due to the way SQLite deals with temporary tables, if you wish to use a
434temporary table in a file-based SQLite database across multiple checkouts
435from the connection pool, such as when using an ORM :class:`.Session` where
436the temporary table should continue to remain after :meth:`.Session.commit` or
437:meth:`.Session.rollback` is called, a pool which maintains a single
438connection must be used. Use :class:`.SingletonThreadPool` if the scope is
439only needed within the current thread, or :class:`.StaticPool` is scope is
440needed within multiple threads for this case::
442 # maintain the same connection per thread
443 from sqlalchemy.pool import SingletonThreadPool
445 engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool)
448 # maintain the same connection across all threads
449 from sqlalchemy.pool import StaticPool
451 engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool)
453Note that :class:`.SingletonThreadPool` should be configured for the number
454of threads that are to be used; beyond that number, connections will be
455closed out in a non deterministic way.
458Dealing with Mixed String / Binary Columns
459------------------------------------------------------
461The SQLite database is weakly typed, and as such it is possible when using
462binary values, which in Python are represented as ``b'some string'``, that a
463particular SQLite database can have data values within different rows where
464some of them will be returned as a ``b''`` value by the Pysqlite driver, and
465others will be returned as Python strings, e.g. ``''`` values. This situation
466is not known to occur if the SQLAlchemy :class:`.LargeBinary` datatype is used
467consistently, however if a particular SQLite database has data that was
468inserted using the Pysqlite driver directly, or when using the SQLAlchemy
469:class:`.String` type which was later changed to :class:`.LargeBinary`, the
470table will not be consistently readable because SQLAlchemy's
471:class:`.LargeBinary` datatype does not handle strings so it has no way of
472"encoding" a value that is in string format.
474To deal with a SQLite table that has mixed string / binary data in the
475same column, use a custom type that will check each row individually::
477 from sqlalchemy import String
478 from sqlalchemy import TypeDecorator
481 class MixedBinary(TypeDecorator):
482 impl = String
483 cache_ok = True
485 def process_result_value(self, value, dialect):
486 if isinstance(value, str):
487 value = bytes(value, "utf-8")
488 elif value is not None:
489 value = bytes(value)
491 return value
493Then use the above ``MixedBinary`` datatype in the place where
494:class:`.LargeBinary` would normally be used.
496.. _pysqlite_serializable:
498Serializable isolation / Savepoints / Transactional DDL
499-------------------------------------------------------
501A newly revised version of this important section is now available
502at the top level of the SQLAlchemy SQLite documentation, in the section
503:ref:`sqlite_transactions`.
506.. _pysqlite_udfs:
508User-Defined Functions
509----------------------
511pysqlite supports a `create_function() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function>`_
512method that allows us to create our own user-defined functions (UDFs) in Python and use them directly in SQLite queries.
513These functions are registered with a specific DBAPI Connection.
515SQLAlchemy uses connection pooling with file-based SQLite databases, so we need to ensure that the UDF is attached to the
516connection when it is created. That is accomplished with an event listener::
518 from sqlalchemy import create_engine
519 from sqlalchemy import event
520 from sqlalchemy import text
523 def udf():
524 return "udf-ok"
527 engine = create_engine("sqlite:///./db_file")
530 @event.listens_for(engine, "connect")
531 def connect(conn, rec):
532 conn.create_function("udf", 0, udf)
535 for i in range(5):
536 with engine.connect() as conn:
537 print(conn.scalar(text("SELECT UDF()")))
539""" # noqa
541from __future__ import annotations
543import math
544import os
545import re
546import sys
547from typing import Any
548from typing import Callable
549from typing import cast
550from typing import Optional
551from typing import Pattern
552from typing import TYPE_CHECKING
553from typing import TypeVar
554from typing import Union
556from .base import DATE
557from .base import DATETIME
558from .base import SQLiteDialect
559from ... import exc
560from ... import pool
561from ... import types as sqltypes
562from ... import util
563from ...util.typing import Self
565if TYPE_CHECKING:
566 from ...engine.interfaces import ConnectArgsType
567 from ...engine.interfaces import DBAPIConnection
568 from ...engine.interfaces import DBAPICursor
569 from ...engine.interfaces import DBAPIModule
570 from ...engine.interfaces import IsolationLevel
571 from ...engine.interfaces import ServerVersionInfoType
572 from ...engine.url import URL
573 from ...pool.base import PoolProxiedConnection
574 from ...sql.type_api import _BindProcessorType
575 from ...sql.type_api import _ResultProcessorType
578class _SQLite_pysqliteTimeStamp(DATETIME):
579 def bind_processor( # type: ignore[override]
580 self, dialect: SQLiteDialect
581 ) -> Optional[_BindProcessorType[Any]]:
582 if dialect.native_datetime:
583 return None
584 else:
585 return DATETIME.bind_processor(self, dialect)
587 def result_processor( # type: ignore[override]
588 self, dialect: SQLiteDialect, coltype: object
589 ) -> Optional[_ResultProcessorType[Any]]:
590 if dialect.native_datetime:
591 return None
592 else:
593 return DATETIME.result_processor(self, dialect, coltype)
596class _SQLite_pysqliteDate(DATE):
597 def bind_processor( # type: ignore[override]
598 self, dialect: SQLiteDialect
599 ) -> Optional[_BindProcessorType[Any]]:
600 if dialect.native_datetime:
601 return None
602 else:
603 return DATE.bind_processor(self, dialect)
605 def result_processor( # type: ignore[override]
606 self, dialect: SQLiteDialect, coltype: object
607 ) -> Optional[_ResultProcessorType[Any]]:
608 if dialect.native_datetime:
609 return None
610 else:
611 return DATE.result_processor(self, dialect, coltype)
614class SQLiteDialect_pysqlite(SQLiteDialect):
615 default_paramstyle = "qmark"
616 supports_statement_cache = True
617 returns_native_bytes = True
619 colspecs = util.update_copy(
620 SQLiteDialect.colspecs,
621 {
622 sqltypes.Date: _SQLite_pysqliteDate,
623 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
624 },
625 )
627 description_encoding = None
629 driver = "pysqlite"
631 @classmethod
632 def import_dbapi(cls) -> DBAPIModule:
633 from sqlite3 import dbapi2 as sqlite
635 return cast("DBAPIModule", sqlite)
637 @classmethod
638 def _is_url_file_db(cls, url: URL) -> bool:
639 if (url.database and url.database != ":memory:") and (
640 url.query.get("mode", None) != "memory"
641 ):
642 return True
643 else:
644 return False
646 @classmethod
647 def _warn_memory_mode_pool_selection(
648 cls,
649 url: URL,
650 current_pool: type[pool.Pool],
651 future_pool: type[pool.Pool],
652 ) -> None:
653 """Warn when the ``mode=memory`` query string argument is what
654 caused a single-connection pool class to be selected.
656 See :ticket:`13433`.
658 """
659 if url.query.get("mode", None) != "memory":
660 return
662 util.warn_deprecated(
663 "Selection of the %s pool class based on the 'mode=memory' "
664 "query string argument is deprecated; a future release will "
665 "use %s for this URL. Indicate the intended pool class "
666 "using the create_engine.poolclass parameter."
667 % (current_pool.__name__, future_pool.__name__),
668 "2.1",
669 code="sqmp",
670 )
672 @classmethod
673 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
674 if cls._is_url_file_db(url):
675 return pool.QueuePool
676 else:
677 cls._warn_memory_mode_pool_selection(
678 url, pool.SingletonThreadPool, pool.QueuePool
679 )
680 return pool.SingletonThreadPool
682 def _get_server_version_info(
683 self, connection: Any
684 ) -> ServerVersionInfoType:
685 return self.dbapi.sqlite_version_info # type: ignore[no-any-return, union-attr] # noqa: E501
687 def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
688 # the ``sqlite3`` module ships with CPython and has no version of
689 # its own (the legacy ``sqlite3.version`` attribute was frozen at
690 # 2.6.0 and removed in Python 3.14), so the Python version is used
691 # here, which is what its feature set actually tracks. The version
692 # of the SQLite library itself is available as
693 # :attr:`.Dialect.server_version_info`.
694 return util.VersionInfo(sys.version_info[:3])
696 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
697 {
698 "AUTOCOMMIT": None,
699 }
700 )
702 def set_isolation_level(
703 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
704 ) -> None:
705 if level == "AUTOCOMMIT":
706 dbapi_connection.isolation_level = None
707 else:
708 dbapi_connection.isolation_level = ""
709 return super().set_isolation_level(dbapi_connection, level)
711 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
712 return dbapi_conn.isolation_level is None
714 def on_connect(self) -> Callable[[DBAPIConnection], None]:
715 def regexp(a: str, b: Optional[str]) -> Optional[bool]:
716 if b is None:
717 return None
718 return re.search(a, b) is not None
720 if self._get_server_version_info(None) >= (3, 9):
721 # sqlite must be greater than 3.8.3 for deterministic=True
722 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
723 # the check is more conservative since there were still issues
724 # with following 3.8 sqlite versions
725 create_func_kw = {"deterministic": True}
726 else:
727 create_func_kw = {}
729 def set_regexp(dbapi_connection: DBAPIConnection) -> None:
730 dbapi_connection.create_function(
731 "regexp", 2, regexp, **create_func_kw
732 )
734 def floor_func(dbapi_connection: DBAPIConnection) -> None:
735 # NOTE: floor is optionally present in sqlite 3.35+ , however
736 # as it is normally non-present we deliver floor() unconditionally
737 # for now.
738 # https://www.sqlite.org/lang_mathfunc.html
739 dbapi_connection.create_function(
740 "floor", 1, math.floor, **create_func_kw
741 )
743 fns = [set_regexp, floor_func]
745 def connect(conn: DBAPIConnection) -> None:
746 for fn in fns:
747 fn(conn)
749 return connect
751 def create_connect_args(self, url: URL) -> ConnectArgsType:
752 if url.username or url.password or url.host or url.port:
753 raise exc.ArgumentError(
754 "Invalid SQLite URL: %s\n"
755 "Valid SQLite URL forms are:\n"
756 " sqlite:///:memory: (or, sqlite://)\n"
757 " sqlite:///relative/path/to/file.db\n"
758 " sqlite:////absolute/path/to/file.db" % (url,)
759 )
761 # theoretically, this list can be augmented, at least as far as
762 # parameter names accepted by sqlite3/pysqlite, using
763 # inspect.getfullargspec(). for the moment this seems like overkill
764 # as these parameters don't change very often, and as always,
765 # parameters passed to connect_args will always go to the
766 # sqlite3/pysqlite driver.
767 pysqlite_args = [
768 ("uri", bool),
769 ("timeout", float),
770 ("isolation_level", str),
771 ("detect_types", int),
772 ("check_same_thread", bool),
773 ("cached_statements", int),
774 ]
775 opts = url.query
776 pysqlite_opts: dict[str, Any] = {}
777 for key, type_ in pysqlite_args:
778 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
780 if pysqlite_opts.get("uri", False):
781 uri_opts = dict(opts)
782 # here, we are actually separating the parameters that go to
783 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
784 # two names conflict? again, this seems to be not the case right
785 # now, and in the case that new names are added to
786 # either side which overlap, again the sqlite3/pysqlite parameters
787 # can be passed through connect_args instead of in the URL.
788 # If SQLite native URIs add a parameter like "timeout" that
789 # we already have listed here for the python driver, then we need
790 # to adjust for that here.
791 for key, type_ in pysqlite_args:
792 uri_opts.pop(key, None)
793 filename: str = url.database # type: ignore[assignment]
794 if uri_opts:
795 # sorting of keys is for unit test support
796 filename += "?" + (
797 "&".join(
798 "%s=%s" % (key, uri_opts[key])
799 for key in sorted(uri_opts)
800 )
801 )
802 else:
803 # without uri=True, the SQLite URI query string is not in
804 # play at all, so anything left over here is silently
805 # discarded; warn rather than have it appear to take effect
806 ignored = sorted(
807 set(opts).difference(key for key, _ in pysqlite_args)
808 )
809 if ignored:
810 util.warn(
811 "Query string argument(s) %s are not accepted by the "
812 "pysqlite driver and are being ignored; SQLite URI "
813 "arguments require that 'uri=true' also be present "
814 "in the URL."
815 % (", ".join("'%s'" % key for key in ignored),),
816 code="squa",
817 )
819 filename = url.database or ":memory:"
820 if filename != ":memory:":
821 filename = os.path.abspath(filename)
823 pysqlite_opts.setdefault(
824 "check_same_thread", not self._is_url_file_db(url)
825 )
827 return ([filename], pysqlite_opts)
829 def is_disconnect(
830 self,
831 e: DBAPIModule.Error,
832 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
833 cursor: Optional[DBAPICursor],
834 ) -> bool:
835 self.dbapi = cast("DBAPIModule", self.dbapi)
836 return isinstance(
837 e, self.dbapi.ProgrammingError
838 ) and "Cannot operate on a closed database." in str(e)
841dialect = SQLiteDialect_pysqlite
844class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
845 """numeric dialect for testing only
847 internal use only. This dialect is **NOT** supported by SQLAlchemy
848 and may change at any time.
850 """
852 supports_statement_cache = True
853 default_paramstyle = "numeric"
854 driver = "pysqlite_numeric"
856 _first_bind = ":1"
857 _not_in_statement_regexp: Optional[Pattern[str]] = None
859 def __init__(self, *arg: Any, **kw: Any) -> None:
860 kw.setdefault("paramstyle", "numeric")
861 super().__init__(*arg, **kw)
863 def create_connect_args(self, url: URL) -> ConnectArgsType:
864 arg, opts = super().create_connect_args(url)
865 opts["factory"] = self._fix_sqlite_issue_99953()
866 return arg, opts
868 def _fix_sqlite_issue_99953(self) -> Any:
869 import sqlite3
871 first_bind = self._first_bind
872 if self._not_in_statement_regexp:
873 nis = self._not_in_statement_regexp
875 def _test_sql(sql: str) -> None:
876 m = nis.search(sql)
877 assert not m, f"Found {nis.pattern!r} in {sql!r}"
879 else:
881 def _test_sql(sql: str) -> None:
882 pass
884 def _numeric_param_as_dict(
885 parameters: Any,
886 ) -> Union[dict[str, Any], tuple[Any, ...]]:
887 if parameters:
888 assert isinstance(parameters, tuple)
889 return {
890 str(idx): value for idx, value in enumerate(parameters, 1)
891 }
892 else:
893 return ()
895 class SQLiteFix99953Cursor(sqlite3.Cursor):
896 def execute(self, sql: str, parameters: Any = ()) -> Self:
897 _test_sql(sql)
898 if first_bind in sql:
899 parameters = _numeric_param_as_dict(parameters)
900 return super().execute(sql, parameters)
902 def executemany(self, sql: str, parameters: Any) -> Self:
903 _test_sql(sql)
904 if first_bind in sql:
905 parameters = [
906 _numeric_param_as_dict(p) for p in parameters
907 ]
908 return super().executemany(sql, parameters)
910 class SQLiteFix99953Connection(sqlite3.Connection):
911 _CursorT = TypeVar("_CursorT", bound=sqlite3.Cursor)
913 def cursor(
914 self,
915 factory: Optional[
916 Callable[[sqlite3.Connection], _CursorT]
917 ] = None,
918 ) -> _CursorT:
919 if factory is None:
920 factory = SQLiteFix99953Cursor # type: ignore[assignment]
921 return super().cursor(factory=factory) # type: ignore[return-value] # noqa[E501]
923 def execute(
924 self, sql: str, parameters: Any = ()
925 ) -> sqlite3.Cursor:
926 _test_sql(sql)
927 if first_bind in sql:
928 parameters = _numeric_param_as_dict(parameters)
929 return super().execute(sql, parameters)
931 def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor:
932 _test_sql(sql)
933 if first_bind in sql:
934 parameters = [
935 _numeric_param_as_dict(p) for p in parameters
936 ]
937 return super().executemany(sql, parameters)
939 return SQLiteFix99953Connection
942class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
943 """numeric dialect that uses $ for testing only
945 internal use only. This dialect is **NOT** supported by SQLAlchemy
946 and may change at any time.
948 """
950 supports_statement_cache = True
951 default_paramstyle = "numeric_dollar"
952 driver = "pysqlite_dollar"
954 _first_bind = "$1"
955 _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
957 def __init__(self, *arg: Any, **kw: Any) -> None:
958 kw.setdefault("paramstyle", "numeric_dollar")
959 super().__init__(*arg, **kw)