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.. versionadded:: 1.3.9
126.. seealso::
128 `Uniform Resource Identifiers <https://www.sqlite.org/uri.html>`_ - in
129 the SQLite documentation
131.. _pysqlite_regexp:
133Regular Expression Support
134---------------------------
136.. versionadded:: 1.4
138Support for the :meth:`_sql.ColumnOperators.regexp_match` operator is provided
139using Python's re.search_ function. SQLite itself does not include a working
140regular expression operator; instead, it includes a non-implemented placeholder
141operator ``REGEXP`` that calls a user-defined function that must be provided.
143SQLAlchemy's implementation makes use of the pysqlite create_function_ hook
144as follows::
147 def regexp(a, b):
148 return re.search(a, b) is not None
151 sqlite_connection.create_function(
152 "regexp",
153 2,
154 regexp,
155 )
157There is currently no support for regular expression flags as a separate
158argument, as these are not supported by SQLite's REGEXP operator, however these
159may be included inline within the regular expression string. See `Python regular expressions`_ for
160details.
162.. seealso::
164 `Python regular expressions`_: Documentation for Python's regular expression syntax.
166.. _create_function: https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
168.. _re.search: https://docs.python.org/3/library/re.html#re.search
170.. _Python regular expressions: https://docs.python.org/3/library/re.html#re.search
174Compatibility with sqlite3 "native" date and datetime types
175-----------------------------------------------------------
177The pysqlite driver includes the sqlite3.PARSE_DECLTYPES and
178sqlite3.PARSE_COLNAMES options, which have the effect of any column
179or expression explicitly cast as "date" or "timestamp" will be converted
180to a Python date or datetime object. The date and datetime types provided
181with the pysqlite dialect are not currently compatible with these options,
182since they render the ISO date/datetime including microseconds, which
183pysqlite's driver does not. Additionally, SQLAlchemy does not at
184this time automatically render the "cast" syntax required for the
185freestanding functions "current_timestamp" and "current_date" to return
186datetime/date types natively. Unfortunately, pysqlite
187does not provide the standard DBAPI types in ``cursor.description``,
188leaving SQLAlchemy with no way to detect these types on the fly
189without expensive per-row type checks.
191Keeping in mind that pysqlite's parsing option is not recommended,
192nor should be necessary, for use with SQLAlchemy, usage of PARSE_DECLTYPES
193can be forced if one configures "native_datetime=True" on create_engine()::
195 engine = create_engine(
196 "sqlite://",
197 connect_args={
198 "detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES
199 },
200 native_datetime=True,
201 )
203With this flag enabled, the DATE and TIMESTAMP types (but note - not the
204DATETIME or TIME types...confused yet ?) will not perform any bind parameter
205or result processing. Execution of "func.current_date()" will return a string.
206"func.current_timestamp()" is registered as returning a DATETIME type in
207SQLAlchemy, so this function still receives SQLAlchemy-level result
208processing.
210.. _pysqlite_threading_pooling:
212Concurrency/Threading/Pooling Behavior
213--------------------------------------
215The ``sqlite3`` DBAPI by default prohibits the use of a particular connection
216in a thread which is not the one in which it was created. As SQLite has
217matured, it's behavior under multiple threads has improved, and even includes
218options for memory only databases to be used in multiple threads.
220The thread prohibition is known as "check same thread" and may be controlled
221using the ``sqlite3`` parameter ``check_same_thread``, which will disable or
222enable this check. SQLAlchemy's default behavior here is to set
223``check_same_thread`` to ``False`` automatically whenever a file-based database
224is in use, to establish compatibility with the default pool class
225:class:`.QueuePool`.
227The SQLAlchemy ``pysqlite`` DBAPI establishes the connection pool differently
228based on the kind of SQLite database that's requested:
230* When a ``:memory:`` SQLite database is specified, the dialect by default
231 will use :class:`.SingletonThreadPool`. This pool maintains a single
232 connection per thread, so that all access to the engine within the current
233 thread use the same ``:memory:`` database - other threads would access a
234 different ``:memory:`` database. The ``check_same_thread`` parameter
235 defaults to ``True``.
236* When a file-based database is specified, the dialect will use
237 :class:`.QueuePool` as the source of connections. at the same time,
238 the ``check_same_thread`` flag is set to False by default unless overridden.
240 .. versionchanged:: 2.0
242 SQLite file database engines now use :class:`.QueuePool` by default.
243 Previously, :class:`.NullPool` were used. The :class:`.NullPool` class
244 may be used by specifying it via the
245 :paramref:`_sa.create_engine.poolclass` parameter.
247Disabling Connection Pooling for File Databases
248^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
250Pooling may be disabled for a file based database by specifying the
251:class:`.NullPool` implementation for the :func:`_sa.create_engine.poolclass`
252parameter::
254 from sqlalchemy import NullPool
256 engine = create_engine("sqlite:///myfile.db", poolclass=NullPool)
258It's been observed that the :class:`.NullPool` implementation incurs an
259extremely small performance overhead for repeated checkouts due to the lack of
260connection reuse implemented by :class:`.QueuePool`. However, it still
261may be beneficial to use this class if the application is experiencing
262issues with files being locked.
264Using a Memory Database in Multiple Threads or Coroutines
265^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
267A ``:memory:`` SQLite database exists only within the scope of a single
268DBAPI connection. It is not possible for two separate ``sqlite3``
269connection objects to access the same ``:memory:`` database unless SQLite's
270shared-cache feature is enabled. Without shared cache, each connection
271creates its own independent in-memory database.
273This means a ``:memory:`` database is **not suitable** for use with
274multiple concurrent threads or coroutines unless either:
276* All workers are fully serialized (mutexed) against each other
277 such that only one is using the database at a time, or
278* SQLite's shared-cache URI feature is used to allow multiple
279 independent connections to access the same in-memory database.
281The same considerations apply when using the :ref:`aiosqlite <aiosqlite>`
282dialect, which wraps ``pysqlite`` connections in an async interface —
283without shared cache, each connection still creates its own independent
284in-memory database.
286The recommended approach for multithreaded or async in-memory
287use is the shared-cache URI feature, described below.
289.. _pysqlite_uri_shared_cache:
291Using a Shared-Cache Memory Database
292~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
294SQLite's
295`URI shared-cache <https://www.sqlite.org/sharedcache.html>`_
296mode allows multiple independent DBAPI connections, each with
297their own transaction state, to access the same in-memory
298database. This is enabled by using a ``file:`` URI with
299``cache=shared`` and passing ``uri=true`` in the query
300string::
302 engine = create_engine("sqlite:///file::memory:?cache=shared&uri=true")
304For async use with aiosqlite, use
305:func:`_asyncio.create_async_engine`::
307 engine = create_async_engine(
308 "sqlite+aiosqlite:///file::memory:?cache=shared&uri=true"
309 )
311Because this URL form is treated as a file-based database by the
312dialect, :class:`.QueuePool` is used automatically and
313``check_same_thread`` defaults to ``False``, so no additional pool
314or connect_args configuration is needed. Each checkout from the
315pool is a distinct DBAPI connection with its own transaction state,
316and the in-memory database persists as long as at least one
317connection remains open.
319The shared-cache database is scoped by the filename component of
320the URI. ``file::memory:`` (empty name) is process-global — all
321engines in the process that use this URI share the same database.
322To maintain multiple independent in-memory databases within the
323same process, supply a distinct name for each::
325 engine_a = create_engine(
326 "sqlite:///file:db_a?mode=memory&cache=shared&uri=true"
327 )
328 engine_b = create_engine(
329 "sqlite:///file:db_b?mode=memory&cache=shared&uri=true"
330 )
332Using StaticPool for Single-Connection Memory Databases
333~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
335An older approach for sharing a ``:memory:`` database among threads
336is to force all checkouts to return the same DBAPI connection using
337:class:`.StaticPool`. **This approach does not support any form of
338concurrency** and is only useful when all access to the engine is
339fully serialized, such as in single-threaded test suites::
341 from sqlalchemy.pool import StaticPool
343 engine = create_engine(
344 "sqlite://",
345 connect_args={"check_same_thread": False},
346 poolclass=StaticPool,
347 )
349.. warning::
351 Because :class:`.StaticPool` maintains a single DBAPI connection,
352 all :class:`.Session` or :class:`.Connection` objects that use
353 this engine share that one underlying connection and its single
354 SQLite transaction state. A ``ROLLBACK`` issued by one session
355 (e.g. during error handling) will also roll back uncommitted work
356 from any other session, and concurrent ``COMMIT`` / ``ROLLBACK``
357 calls can interfere with each other unpredictably. This approach
358 is only appropriate when access to the engine is fully serialized,
359 such as in single-threaded test suites. For concurrent workloads,
360 use the shared-cache URI approach described above.
362Using Temporary Tables with SQLite
363^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
365Due to the way SQLite deals with temporary tables, if you wish to use a
366temporary table in a file-based SQLite database across multiple checkouts
367from the connection pool, such as when using an ORM :class:`.Session` where
368the temporary table should continue to remain after :meth:`.Session.commit` or
369:meth:`.Session.rollback` is called, a pool which maintains a single
370connection must be used. Use :class:`.SingletonThreadPool` if the scope is
371only needed within the current thread, or :class:`.StaticPool` is scope is
372needed within multiple threads for this case::
374 # maintain the same connection per thread
375 from sqlalchemy.pool import SingletonThreadPool
377 engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool)
380 # maintain the same connection across all threads
381 from sqlalchemy.pool import StaticPool
383 engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool)
385Note that :class:`.SingletonThreadPool` should be configured for the number
386of threads that are to be used; beyond that number, connections will be
387closed out in a non deterministic way.
390Dealing with Mixed String / Binary Columns
391------------------------------------------------------
393The SQLite database is weakly typed, and as such it is possible when using
394binary values, which in Python are represented as ``b'some string'``, that a
395particular SQLite database can have data values within different rows where
396some of them will be returned as a ``b''`` value by the Pysqlite driver, and
397others will be returned as Python strings, e.g. ``''`` values. This situation
398is not known to occur if the SQLAlchemy :class:`.LargeBinary` datatype is used
399consistently, however if a particular SQLite database has data that was
400inserted using the Pysqlite driver directly, or when using the SQLAlchemy
401:class:`.String` type which was later changed to :class:`.LargeBinary`, the
402table will not be consistently readable because SQLAlchemy's
403:class:`.LargeBinary` datatype does not handle strings so it has no way of
404"encoding" a value that is in string format.
406To deal with a SQLite table that has mixed string / binary data in the
407same column, use a custom type that will check each row individually::
409 from sqlalchemy import String
410 from sqlalchemy import TypeDecorator
413 class MixedBinary(TypeDecorator):
414 impl = String
415 cache_ok = True
417 def process_result_value(self, value, dialect):
418 if isinstance(value, str):
419 value = bytes(value, "utf-8")
420 elif value is not None:
421 value = bytes(value)
423 return value
425Then use the above ``MixedBinary`` datatype in the place where
426:class:`.LargeBinary` would normally be used.
428.. _pysqlite_serializable:
430Serializable isolation / Savepoints / Transactional DDL
431-------------------------------------------------------
433A newly revised version of this important section is now available
434at the top level of the SQLAlchemy SQLite documentation, in the section
435:ref:`sqlite_transactions`.
438.. _pysqlite_udfs:
440User-Defined Functions
441----------------------
443pysqlite supports a `create_function() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function>`_
444method that allows us to create our own user-defined functions (UDFs) in Python and use them directly in SQLite queries.
445These functions are registered with a specific DBAPI Connection.
447SQLAlchemy uses connection pooling with file-based SQLite databases, so we need to ensure that the UDF is attached to the
448connection when it is created. That is accomplished with an event listener::
450 from sqlalchemy import create_engine
451 from sqlalchemy import event
452 from sqlalchemy import text
455 def udf():
456 return "udf-ok"
459 engine = create_engine("sqlite:///./db_file")
462 @event.listens_for(engine, "connect")
463 def connect(conn, rec):
464 conn.create_function("udf", 0, udf)
467 for i in range(5):
468 with engine.connect() as conn:
469 print(conn.scalar(text("SELECT UDF()")))
471""" # noqa
473from __future__ import annotations
475import math
476import os
477import re
478from typing import Any
479from typing import Callable
480from typing import cast
481from typing import Optional
482from typing import Pattern
483from typing import TYPE_CHECKING
484from typing import TypeVar
485from typing import Union
487from .base import DATE
488from .base import DATETIME
489from .base import SQLiteDialect
490from ... import exc
491from ... import pool
492from ... import types as sqltypes
493from ... import util
494from ...util.typing import Self
496if TYPE_CHECKING:
497 from ...engine.interfaces import ConnectArgsType
498 from ...engine.interfaces import DBAPIConnection
499 from ...engine.interfaces import DBAPICursor
500 from ...engine.interfaces import DBAPIModule
501 from ...engine.interfaces import IsolationLevel
502 from ...engine.interfaces import VersionInfoType
503 from ...engine.url import URL
504 from ...pool.base import PoolProxiedConnection
505 from ...sql.type_api import _BindProcessorType
506 from ...sql.type_api import _ResultProcessorType
509class _SQLite_pysqliteTimeStamp(DATETIME):
510 def bind_processor( # type: ignore[override]
511 self, dialect: SQLiteDialect
512 ) -> Optional[_BindProcessorType[Any]]:
513 if dialect.native_datetime:
514 return None
515 else:
516 return DATETIME.bind_processor(self, dialect)
518 def result_processor( # type: ignore[override]
519 self, dialect: SQLiteDialect, coltype: object
520 ) -> Optional[_ResultProcessorType[Any]]:
521 if dialect.native_datetime:
522 return None
523 else:
524 return DATETIME.result_processor(self, dialect, coltype)
527class _SQLite_pysqliteDate(DATE):
528 def bind_processor( # type: ignore[override]
529 self, dialect: SQLiteDialect
530 ) -> Optional[_BindProcessorType[Any]]:
531 if dialect.native_datetime:
532 return None
533 else:
534 return DATE.bind_processor(self, dialect)
536 def result_processor( # type: ignore[override]
537 self, dialect: SQLiteDialect, coltype: object
538 ) -> Optional[_ResultProcessorType[Any]]:
539 if dialect.native_datetime:
540 return None
541 else:
542 return DATE.result_processor(self, dialect, coltype)
545class SQLiteDialect_pysqlite(SQLiteDialect):
546 default_paramstyle = "qmark"
547 supports_statement_cache = True
548 returns_native_bytes = True
550 colspecs = util.update_copy(
551 SQLiteDialect.colspecs,
552 {
553 sqltypes.Date: _SQLite_pysqliteDate,
554 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
555 },
556 )
558 description_encoding = None
560 driver = "pysqlite"
562 @classmethod
563 def import_dbapi(cls) -> DBAPIModule:
564 from sqlite3 import dbapi2 as sqlite
566 return cast("DBAPIModule", sqlite)
568 @classmethod
569 def _is_url_file_db(cls, url: URL) -> bool:
570 if (url.database and url.database != ":memory:") and (
571 url.query.get("mode", None) != "memory"
572 ):
573 return True
574 else:
575 return False
577 @classmethod
578 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
579 if cls._is_url_file_db(url):
580 return pool.QueuePool
581 else:
582 return pool.SingletonThreadPool
584 def _get_server_version_info(self, connection: Any) -> VersionInfoType:
585 return self.dbapi.sqlite_version_info # type: ignore
587 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
588 {
589 "AUTOCOMMIT": None, # type: ignore[dict-item]
590 }
591 )
593 def set_isolation_level(
594 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
595 ) -> None:
596 if level == "AUTOCOMMIT":
597 dbapi_connection.isolation_level = None
598 else:
599 dbapi_connection.isolation_level = ""
600 return super().set_isolation_level(dbapi_connection, level)
602 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
603 return dbapi_conn.isolation_level is None
605 def on_connect(self) -> Callable[[DBAPIConnection], None]:
606 def regexp(a: str, b: Optional[str]) -> Optional[bool]:
607 if b is None:
608 return None
609 return re.search(a, b) is not None
611 if util.py38 and self._get_server_version_info(None) >= (3, 9):
612 # sqlite must be greater than 3.8.3 for deterministic=True
613 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
614 # the check is more conservative since there were still issues
615 # with following 3.8 sqlite versions
616 create_func_kw = {"deterministic": True}
617 else:
618 create_func_kw = {}
620 def set_regexp(dbapi_connection: DBAPIConnection) -> None:
621 dbapi_connection.create_function(
622 "regexp", 2, regexp, **create_func_kw
623 )
625 def floor_func(dbapi_connection: DBAPIConnection) -> None:
626 # NOTE: floor is optionally present in sqlite 3.35+ , however
627 # as it is normally non-present we deliver floor() unconditionally
628 # for now.
629 # https://www.sqlite.org/lang_mathfunc.html
630 dbapi_connection.create_function(
631 "floor", 1, math.floor, **create_func_kw
632 )
634 fns = [set_regexp, floor_func]
636 def connect(conn: DBAPIConnection) -> None:
637 for fn in fns:
638 fn(conn)
640 return connect
642 def create_connect_args(self, url: URL) -> ConnectArgsType:
643 if url.username or url.password or url.host or url.port:
644 raise exc.ArgumentError(
645 "Invalid SQLite URL: %s\n"
646 "Valid SQLite URL forms are:\n"
647 " sqlite:///:memory: (or, sqlite://)\n"
648 " sqlite:///relative/path/to/file.db\n"
649 " sqlite:////absolute/path/to/file.db" % (url,)
650 )
652 # theoretically, this list can be augmented, at least as far as
653 # parameter names accepted by sqlite3/pysqlite, using
654 # inspect.getfullargspec(). for the moment this seems like overkill
655 # as these parameters don't change very often, and as always,
656 # parameters passed to connect_args will always go to the
657 # sqlite3/pysqlite driver.
658 pysqlite_args = [
659 ("uri", bool),
660 ("timeout", float),
661 ("isolation_level", str),
662 ("detect_types", int),
663 ("check_same_thread", bool),
664 ("cached_statements", int),
665 ]
666 opts = url.query
667 pysqlite_opts: dict[str, Any] = {}
668 for key, type_ in pysqlite_args:
669 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
671 if pysqlite_opts.get("uri", False):
672 uri_opts = dict(opts)
673 # here, we are actually separating the parameters that go to
674 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
675 # two names conflict? again, this seems to be not the case right
676 # now, and in the case that new names are added to
677 # either side which overlap, again the sqlite3/pysqlite parameters
678 # can be passed through connect_args instead of in the URL.
679 # If SQLite native URIs add a parameter like "timeout" that
680 # we already have listed here for the python driver, then we need
681 # to adjust for that here.
682 for key, type_ in pysqlite_args:
683 uri_opts.pop(key, None)
684 filename: str = url.database # type: ignore[assignment]
685 if uri_opts:
686 # sorting of keys is for unit test support
687 filename += "?" + (
688 "&".join(
689 "%s=%s" % (key, uri_opts[key])
690 for key in sorted(uri_opts)
691 )
692 )
693 else:
694 filename = url.database or ":memory:"
695 if filename != ":memory:":
696 filename = os.path.abspath(filename)
698 pysqlite_opts.setdefault(
699 "check_same_thread", not self._is_url_file_db(url)
700 )
702 return ([filename], pysqlite_opts)
704 def is_disconnect(
705 self,
706 e: DBAPIModule.Error,
707 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
708 cursor: Optional[DBAPICursor],
709 ) -> bool:
710 self.dbapi = cast("DBAPIModule", self.dbapi)
711 return isinstance(
712 e, self.dbapi.ProgrammingError
713 ) and "Cannot operate on a closed database." in str(e)
716dialect = SQLiteDialect_pysqlite
719class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
720 """numeric dialect for testing only
722 internal use only. This dialect is **NOT** supported by SQLAlchemy
723 and may change at any time.
725 """
727 supports_statement_cache = True
728 default_paramstyle = "numeric"
729 driver = "pysqlite_numeric"
731 _first_bind = ":1"
732 _not_in_statement_regexp: Optional[Pattern[str]] = None
734 def __init__(self, *arg: Any, **kw: Any) -> None:
735 kw.setdefault("paramstyle", "numeric")
736 super().__init__(*arg, **kw)
738 def create_connect_args(self, url: URL) -> ConnectArgsType:
739 arg, opts = super().create_connect_args(url)
740 opts["factory"] = self._fix_sqlite_issue_99953()
741 return arg, opts
743 def _fix_sqlite_issue_99953(self) -> Any:
744 import sqlite3
746 first_bind = self._first_bind
747 if self._not_in_statement_regexp:
748 nis = self._not_in_statement_regexp
750 def _test_sql(sql: str) -> None:
751 m = nis.search(sql)
752 assert not m, f"Found {nis.pattern!r} in {sql!r}"
754 else:
756 def _test_sql(sql: str) -> None:
757 pass
759 def _numeric_param_as_dict(
760 parameters: Any,
761 ) -> Union[dict[str, Any], tuple[Any, ...]]:
762 if parameters:
763 assert isinstance(parameters, tuple)
764 return {
765 str(idx): value for idx, value in enumerate(parameters, 1)
766 }
767 else:
768 return ()
770 class SQLiteFix99953Cursor(sqlite3.Cursor):
771 def execute(self, sql: str, parameters: Any = ()) -> Self:
772 _test_sql(sql)
773 if first_bind in sql:
774 parameters = _numeric_param_as_dict(parameters)
775 return super().execute(sql, parameters)
777 def executemany(self, sql: str, parameters: Any) -> Self:
778 _test_sql(sql)
779 if first_bind in sql:
780 parameters = [
781 _numeric_param_as_dict(p) for p in parameters
782 ]
783 return super().executemany(sql, parameters)
785 class SQLiteFix99953Connection(sqlite3.Connection):
786 _CursorT = TypeVar("_CursorT", bound=sqlite3.Cursor)
788 def cursor(
789 self,
790 factory: Optional[
791 Callable[[sqlite3.Connection], _CursorT]
792 ] = None,
793 ) -> _CursorT:
794 if factory is None:
795 factory = SQLiteFix99953Cursor # type: ignore[assignment]
796 return super().cursor(factory=factory) # type: ignore[return-value] # noqa[E501]
798 def execute(
799 self, sql: str, parameters: Any = ()
800 ) -> sqlite3.Cursor:
801 _test_sql(sql)
802 if first_bind in sql:
803 parameters = _numeric_param_as_dict(parameters)
804 return super().execute(sql, parameters)
806 def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor:
807 _test_sql(sql)
808 if first_bind in sql:
809 parameters = [
810 _numeric_param_as_dict(p) for p in parameters
811 ]
812 return super().executemany(sql, parameters)
814 return SQLiteFix99953Connection
817class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
818 """numeric dialect that uses $ for testing only
820 internal use only. This dialect is **NOT** supported by SQLAlchemy
821 and may change at any time.
823 """
825 supports_statement_cache = True
826 default_paramstyle = "numeric_dollar"
827 driver = "pysqlite_dollar"
829 _first_bind = "$1"
830 _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
832 def __init__(self, *arg: Any, **kw: Any) -> None:
833 kw.setdefault("paramstyle", "numeric_dollar")
834 super().__init__(*arg, **kw)