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.
245Disabling Connection Pooling for File Databases
246^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
248Pooling may be disabled for a file based database by specifying the
249:class:`.NullPool` implementation for the :func:`_sa.create_engine.poolclass`
250parameter::
252 from sqlalchemy import NullPool
254 engine = create_engine("sqlite:///myfile.db", poolclass=NullPool)
256It's been observed that the :class:`.NullPool` implementation incurs an
257extremely small performance overhead for repeated checkouts due to the lack of
258connection reuse implemented by :class:`.QueuePool`. However, it still
259may be beneficial to use this class if the application is experiencing
260issues with files being locked.
262Using a Memory Database in Multiple Threads or Coroutines
263^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
265A ``:memory:`` SQLite database exists only within the scope of a single
266DBAPI connection. It is not possible for two separate ``sqlite3``
267connection objects to access the same ``:memory:`` database unless SQLite's
268shared-cache feature is enabled. Without shared cache, each connection
269creates its own independent in-memory database.
271This means a ``:memory:`` database is **not suitable** for use with
272multiple concurrent threads or coroutines unless either:
274* All workers are fully serialized (mutexed) against each other
275 such that only one is using the database at a time, or
276* SQLite's shared-cache URI feature is used to allow multiple
277 independent connections to access the same in-memory database.
279The same considerations apply when using the :ref:`aiosqlite <aiosqlite>`
280dialect, which wraps ``pysqlite`` connections in an async interface —
281without shared cache, each connection still creates its own independent
282in-memory database.
284The recommended approach for multithreaded or async in-memory
285use is the shared-cache URI feature, described below.
287.. _pysqlite_uri_shared_cache:
289Using a Shared-Cache Memory Database
290~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
292SQLite's
293`URI shared-cache <https://www.sqlite.org/sharedcache.html>`_
294mode allows multiple independent DBAPI connections, each with
295their own transaction state, to access the same in-memory
296database. This is enabled by using a ``file:`` URI with
297``cache=shared`` and passing ``uri=true`` in the query
298string::
300 engine = create_engine("sqlite:///file::memory:?cache=shared&uri=true")
302For async use with aiosqlite, use
303:func:`_asyncio.create_async_engine`::
305 engine = create_async_engine(
306 "sqlite+aiosqlite:///file::memory:?cache=shared&uri=true"
307 )
309Because this URL form is treated as a file-based database by the
310dialect, :class:`.QueuePool` is used automatically and
311``check_same_thread`` defaults to ``False``, so no additional pool
312or connect_args configuration is needed. Each checkout from the
313pool is a distinct DBAPI connection with its own transaction state,
314and the in-memory database persists as long as at least one
315connection remains open.
317The shared-cache database is scoped by the filename component of
318the URI. ``file::memory:`` (empty name) is process-global — all
319engines in the process that use this URI share the same database.
320To maintain multiple independent in-memory databases within the
321same process, supply a distinct name for each::
323 engine_a = create_engine(
324 "sqlite:///file:db_a?mode=memory&cache=shared&uri=true"
325 )
326 engine_b = create_engine(
327 "sqlite:///file:db_b?mode=memory&cache=shared&uri=true"
328 )
330Using StaticPool for Single-Connection Memory Databases
331~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
333An older approach for sharing a ``:memory:`` database among threads
334is to force all checkouts to return the same DBAPI connection using
335:class:`.StaticPool`. **This approach does not support any form of
336concurrency** and is only useful when all access to the engine is
337fully serialized, such as in single-threaded test suites::
339 from sqlalchemy.pool import StaticPool
341 engine = create_engine(
342 "sqlite://",
343 connect_args={"check_same_thread": False},
344 poolclass=StaticPool,
345 )
347.. warning::
349 Because :class:`.StaticPool` maintains a single DBAPI connection,
350 all :class:`.Session` or :class:`.Connection` objects that use
351 this engine share that one underlying connection and its single
352 SQLite transaction state. A ``ROLLBACK`` issued by one session
353 (e.g. during error handling) will also roll back uncommitted work
354 from any other session, and concurrent ``COMMIT`` / ``ROLLBACK``
355 calls can interfere with each other unpredictably. This approach
356 is only appropriate when access to the engine is fully serialized,
357 such as in single-threaded test suites. For concurrent workloads,
358 use the shared-cache URI approach described above.
360Using Temporary Tables with SQLite
361^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
363Due to the way SQLite deals with temporary tables, if you wish to use a
364temporary table in a file-based SQLite database across multiple checkouts
365from the connection pool, such as when using an ORM :class:`.Session` where
366the temporary table should continue to remain after :meth:`.Session.commit` or
367:meth:`.Session.rollback` is called, a pool which maintains a single
368connection must be used. Use :class:`.SingletonThreadPool` if the scope is
369only needed within the current thread, or :class:`.StaticPool` is scope is
370needed within multiple threads for this case::
372 # maintain the same connection per thread
373 from sqlalchemy.pool import SingletonThreadPool
375 engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool)
378 # maintain the same connection across all threads
379 from sqlalchemy.pool import StaticPool
381 engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool)
383Note that :class:`.SingletonThreadPool` should be configured for the number
384of threads that are to be used; beyond that number, connections will be
385closed out in a non deterministic way.
388Dealing with Mixed String / Binary Columns
389------------------------------------------------------
391The SQLite database is weakly typed, and as such it is possible when using
392binary values, which in Python are represented as ``b'some string'``, that a
393particular SQLite database can have data values within different rows where
394some of them will be returned as a ``b''`` value by the Pysqlite driver, and
395others will be returned as Python strings, e.g. ``''`` values. This situation
396is not known to occur if the SQLAlchemy :class:`.LargeBinary` datatype is used
397consistently, however if a particular SQLite database has data that was
398inserted using the Pysqlite driver directly, or when using the SQLAlchemy
399:class:`.String` type which was later changed to :class:`.LargeBinary`, the
400table will not be consistently readable because SQLAlchemy's
401:class:`.LargeBinary` datatype does not handle strings so it has no way of
402"encoding" a value that is in string format.
404To deal with a SQLite table that has mixed string / binary data in the
405same column, use a custom type that will check each row individually::
407 from sqlalchemy import String
408 from sqlalchemy import TypeDecorator
411 class MixedBinary(TypeDecorator):
412 impl = String
413 cache_ok = True
415 def process_result_value(self, value, dialect):
416 if isinstance(value, str):
417 value = bytes(value, "utf-8")
418 elif value is not None:
419 value = bytes(value)
421 return value
423Then use the above ``MixedBinary`` datatype in the place where
424:class:`.LargeBinary` would normally be used.
426.. _pysqlite_serializable:
428Serializable isolation / Savepoints / Transactional DDL
429-------------------------------------------------------
431A newly revised version of this important section is now available
432at the top level of the SQLAlchemy SQLite documentation, in the section
433:ref:`sqlite_transactions`.
436.. _pysqlite_udfs:
438User-Defined Functions
439----------------------
441pysqlite supports a `create_function() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function>`_
442method that allows us to create our own user-defined functions (UDFs) in Python and use them directly in SQLite queries.
443These functions are registered with a specific DBAPI Connection.
445SQLAlchemy uses connection pooling with file-based SQLite databases, so we need to ensure that the UDF is attached to the
446connection when it is created. That is accomplished with an event listener::
448 from sqlalchemy import create_engine
449 from sqlalchemy import event
450 from sqlalchemy import text
453 def udf():
454 return "udf-ok"
457 engine = create_engine("sqlite:///./db_file")
460 @event.listens_for(engine, "connect")
461 def connect(conn, rec):
462 conn.create_function("udf", 0, udf)
465 for i in range(5):
466 with engine.connect() as conn:
467 print(conn.scalar(text("SELECT UDF()")))
469""" # noqa
471from __future__ import annotations
473import math
474import os
475import re
476import sys
477from typing import Any
478from typing import Callable
479from typing import cast
480from typing import Optional
481from typing import Pattern
482from typing import TYPE_CHECKING
483from typing import TypeVar
484from typing import Union
486from .base import DATE
487from .base import DATETIME
488from .base import SQLiteDialect
489from ... import exc
490from ... import pool
491from ... import types as sqltypes
492from ... import util
493from ...util.typing import Self
495if TYPE_CHECKING:
496 from ...engine.interfaces import ConnectArgsType
497 from ...engine.interfaces import DBAPIConnection
498 from ...engine.interfaces import DBAPICursor
499 from ...engine.interfaces import DBAPIModule
500 from ...engine.interfaces import IsolationLevel
501 from ...engine.interfaces import ServerVersionInfoType
502 from ...engine.url import URL
503 from ...pool.base import PoolProxiedConnection
504 from ...sql.type_api import _BindProcessorType
505 from ...sql.type_api import _ResultProcessorType
508class _SQLite_pysqliteTimeStamp(DATETIME):
509 def bind_processor( # type: ignore[override]
510 self, dialect: SQLiteDialect
511 ) -> Optional[_BindProcessorType[Any]]:
512 if dialect.native_datetime:
513 return None
514 else:
515 return DATETIME.bind_processor(self, dialect)
517 def result_processor( # type: ignore[override]
518 self, dialect: SQLiteDialect, coltype: object
519 ) -> Optional[_ResultProcessorType[Any]]:
520 if dialect.native_datetime:
521 return None
522 else:
523 return DATETIME.result_processor(self, dialect, coltype)
526class _SQLite_pysqliteDate(DATE):
527 def bind_processor( # type: ignore[override]
528 self, dialect: SQLiteDialect
529 ) -> Optional[_BindProcessorType[Any]]:
530 if dialect.native_datetime:
531 return None
532 else:
533 return DATE.bind_processor(self, dialect)
535 def result_processor( # type: ignore[override]
536 self, dialect: SQLiteDialect, coltype: object
537 ) -> Optional[_ResultProcessorType[Any]]:
538 if dialect.native_datetime:
539 return None
540 else:
541 return DATE.result_processor(self, dialect, coltype)
544class SQLiteDialect_pysqlite(SQLiteDialect):
545 default_paramstyle = "qmark"
546 supports_statement_cache = True
547 returns_native_bytes = True
549 colspecs = util.update_copy(
550 SQLiteDialect.colspecs,
551 {
552 sqltypes.Date: _SQLite_pysqliteDate,
553 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
554 },
555 )
557 description_encoding = None
559 driver = "pysqlite"
561 @classmethod
562 def import_dbapi(cls) -> DBAPIModule:
563 from sqlite3 import dbapi2 as sqlite
565 return cast("DBAPIModule", sqlite)
567 @classmethod
568 def _is_url_file_db(cls, url: URL) -> bool:
569 if (url.database and url.database != ":memory:") and (
570 url.query.get("mode", None) != "memory"
571 ):
572 return True
573 else:
574 return False
576 @classmethod
577 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
578 if cls._is_url_file_db(url):
579 return pool.QueuePool
580 else:
581 return pool.SingletonThreadPool
583 def _get_server_version_info(
584 self, connection: Any
585 ) -> ServerVersionInfoType:
586 return self.dbapi.sqlite_version_info # type: ignore[no-any-return, union-attr] # noqa: E501
588 def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo:
589 # the ``sqlite3`` module ships with CPython and has no version of
590 # its own (the legacy ``sqlite3.version`` attribute was frozen at
591 # 2.6.0 and removed in Python 3.14), so the Python version is used
592 # here, which is what its feature set actually tracks. The version
593 # of the SQLite library itself is available as
594 # :attr:`.Dialect.server_version_info`.
595 return util.VersionInfo(sys.version_info[:3])
597 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
598 {
599 "AUTOCOMMIT": None,
600 }
601 )
603 def set_isolation_level(
604 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
605 ) -> None:
606 if level == "AUTOCOMMIT":
607 dbapi_connection.isolation_level = None
608 else:
609 dbapi_connection.isolation_level = ""
610 return super().set_isolation_level(dbapi_connection, level)
612 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
613 return dbapi_conn.isolation_level is None
615 def on_connect(self) -> Callable[[DBAPIConnection], None]:
616 def regexp(a: str, b: Optional[str]) -> Optional[bool]:
617 if b is None:
618 return None
619 return re.search(a, b) is not None
621 if self._get_server_version_info(None) >= (3, 9):
622 # sqlite must be greater than 3.8.3 for deterministic=True
623 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
624 # the check is more conservative since there were still issues
625 # with following 3.8 sqlite versions
626 create_func_kw = {"deterministic": True}
627 else:
628 create_func_kw = {}
630 def set_regexp(dbapi_connection: DBAPIConnection) -> None:
631 dbapi_connection.create_function(
632 "regexp", 2, regexp, **create_func_kw
633 )
635 def floor_func(dbapi_connection: DBAPIConnection) -> None:
636 # NOTE: floor is optionally present in sqlite 3.35+ , however
637 # as it is normally non-present we deliver floor() unconditionally
638 # for now.
639 # https://www.sqlite.org/lang_mathfunc.html
640 dbapi_connection.create_function(
641 "floor", 1, math.floor, **create_func_kw
642 )
644 fns = [set_regexp, floor_func]
646 def connect(conn: DBAPIConnection) -> None:
647 for fn in fns:
648 fn(conn)
650 return connect
652 def create_connect_args(self, url: URL) -> ConnectArgsType:
653 if url.username or url.password or url.host or url.port:
654 raise exc.ArgumentError(
655 "Invalid SQLite URL: %s\n"
656 "Valid SQLite URL forms are:\n"
657 " sqlite:///:memory: (or, sqlite://)\n"
658 " sqlite:///relative/path/to/file.db\n"
659 " sqlite:////absolute/path/to/file.db" % (url,)
660 )
662 # theoretically, this list can be augmented, at least as far as
663 # parameter names accepted by sqlite3/pysqlite, using
664 # inspect.getfullargspec(). for the moment this seems like overkill
665 # as these parameters don't change very often, and as always,
666 # parameters passed to connect_args will always go to the
667 # sqlite3/pysqlite driver.
668 pysqlite_args = [
669 ("uri", bool),
670 ("timeout", float),
671 ("isolation_level", str),
672 ("detect_types", int),
673 ("check_same_thread", bool),
674 ("cached_statements", int),
675 ]
676 opts = url.query
677 pysqlite_opts: dict[str, Any] = {}
678 for key, type_ in pysqlite_args:
679 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
681 if pysqlite_opts.get("uri", False):
682 uri_opts = dict(opts)
683 # here, we are actually separating the parameters that go to
684 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
685 # two names conflict? again, this seems to be not the case right
686 # now, and in the case that new names are added to
687 # either side which overlap, again the sqlite3/pysqlite parameters
688 # can be passed through connect_args instead of in the URL.
689 # If SQLite native URIs add a parameter like "timeout" that
690 # we already have listed here for the python driver, then we need
691 # to adjust for that here.
692 for key, type_ in pysqlite_args:
693 uri_opts.pop(key, None)
694 filename: str = url.database # type: ignore[assignment]
695 if uri_opts:
696 # sorting of keys is for unit test support
697 filename += "?" + (
698 "&".join(
699 "%s=%s" % (key, uri_opts[key])
700 for key in sorted(uri_opts)
701 )
702 )
703 else:
704 filename = url.database or ":memory:"
705 if filename != ":memory:":
706 filename = os.path.abspath(filename)
708 pysqlite_opts.setdefault(
709 "check_same_thread", not self._is_url_file_db(url)
710 )
712 return ([filename], pysqlite_opts)
714 def is_disconnect(
715 self,
716 e: DBAPIModule.Error,
717 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
718 cursor: Optional[DBAPICursor],
719 ) -> bool:
720 self.dbapi = cast("DBAPIModule", self.dbapi)
721 return isinstance(
722 e, self.dbapi.ProgrammingError
723 ) and "Cannot operate on a closed database." in str(e)
726dialect = SQLiteDialect_pysqlite
729class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
730 """numeric dialect for testing only
732 internal use only. This dialect is **NOT** supported by SQLAlchemy
733 and may change at any time.
735 """
737 supports_statement_cache = True
738 default_paramstyle = "numeric"
739 driver = "pysqlite_numeric"
741 _first_bind = ":1"
742 _not_in_statement_regexp: Optional[Pattern[str]] = None
744 def __init__(self, *arg: Any, **kw: Any) -> None:
745 kw.setdefault("paramstyle", "numeric")
746 super().__init__(*arg, **kw)
748 def create_connect_args(self, url: URL) -> ConnectArgsType:
749 arg, opts = super().create_connect_args(url)
750 opts["factory"] = self._fix_sqlite_issue_99953()
751 return arg, opts
753 def _fix_sqlite_issue_99953(self) -> Any:
754 import sqlite3
756 first_bind = self._first_bind
757 if self._not_in_statement_regexp:
758 nis = self._not_in_statement_regexp
760 def _test_sql(sql: str) -> None:
761 m = nis.search(sql)
762 assert not m, f"Found {nis.pattern!r} in {sql!r}"
764 else:
766 def _test_sql(sql: str) -> None:
767 pass
769 def _numeric_param_as_dict(
770 parameters: Any,
771 ) -> Union[dict[str, Any], tuple[Any, ...]]:
772 if parameters:
773 assert isinstance(parameters, tuple)
774 return {
775 str(idx): value for idx, value in enumerate(parameters, 1)
776 }
777 else:
778 return ()
780 class SQLiteFix99953Cursor(sqlite3.Cursor):
781 def execute(self, sql: str, parameters: Any = ()) -> Self:
782 _test_sql(sql)
783 if first_bind in sql:
784 parameters = _numeric_param_as_dict(parameters)
785 return super().execute(sql, parameters)
787 def executemany(self, sql: str, parameters: Any) -> Self:
788 _test_sql(sql)
789 if first_bind in sql:
790 parameters = [
791 _numeric_param_as_dict(p) for p in parameters
792 ]
793 return super().executemany(sql, parameters)
795 class SQLiteFix99953Connection(sqlite3.Connection):
796 _CursorT = TypeVar("_CursorT", bound=sqlite3.Cursor)
798 def cursor(
799 self,
800 factory: Optional[
801 Callable[[sqlite3.Connection], _CursorT]
802 ] = None,
803 ) -> _CursorT:
804 if factory is None:
805 factory = SQLiteFix99953Cursor # type: ignore[assignment]
806 return super().cursor(factory=factory) # type: ignore[return-value] # noqa[E501]
808 def execute(
809 self, sql: str, parameters: Any = ()
810 ) -> sqlite3.Cursor:
811 _test_sql(sql)
812 if first_bind in sql:
813 parameters = _numeric_param_as_dict(parameters)
814 return super().execute(sql, parameters)
816 def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor:
817 _test_sql(sql)
818 if first_bind in sql:
819 parameters = [
820 _numeric_param_as_dict(p) for p in parameters
821 ]
822 return super().executemany(sql, parameters)
824 return SQLiteFix99953Connection
827class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
828 """numeric dialect that uses $ for testing only
830 internal use only. This dialect is **NOT** supported by SQLAlchemy
831 and may change at any time.
833 """
835 supports_statement_cache = True
836 default_paramstyle = "numeric_dollar"
837 driver = "pysqlite_dollar"
839 _first_bind = "$1"
840 _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
842 def __init__(self, *arg: Any, **kw: Any) -> None:
843 kw.setdefault("paramstyle", "numeric_dollar")
844 super().__init__(*arg, **kw)