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:
210Threading/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
263^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
265To use a ``:memory:`` database in a multithreaded scenario, the same
266connection object must be shared among threads, since the database exists
267only within the scope of that connection. The
268:class:`.StaticPool` implementation will maintain a single connection
269globally, and the ``check_same_thread`` flag can be passed to Pysqlite
270as ``False``::
272 from sqlalchemy.pool import StaticPool
274 engine = create_engine(
275 "sqlite://",
276 connect_args={"check_same_thread": False},
277 poolclass=StaticPool,
278 )
280Note that using a ``:memory:`` database in multiple threads requires a recent
281version of SQLite.
283Using Temporary Tables with SQLite
284^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
286Due to the way SQLite deals with temporary tables, if you wish to use a
287temporary table in a file-based SQLite database across multiple checkouts
288from the connection pool, such as when using an ORM :class:`.Session` where
289the temporary table should continue to remain after :meth:`.Session.commit` or
290:meth:`.Session.rollback` is called, a pool which maintains a single
291connection must be used. Use :class:`.SingletonThreadPool` if the scope is
292only needed within the current thread, or :class:`.StaticPool` is scope is
293needed within multiple threads for this case::
295 # maintain the same connection per thread
296 from sqlalchemy.pool import SingletonThreadPool
298 engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool)
301 # maintain the same connection across all threads
302 from sqlalchemy.pool import StaticPool
304 engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool)
306Note that :class:`.SingletonThreadPool` should be configured for the number
307of threads that are to be used; beyond that number, connections will be
308closed out in a non deterministic way.
311Dealing with Mixed String / Binary Columns
312------------------------------------------------------
314The SQLite database is weakly typed, and as such it is possible when using
315binary values, which in Python are represented as ``b'some string'``, that a
316particular SQLite database can have data values within different rows where
317some of them will be returned as a ``b''`` value by the Pysqlite driver, and
318others will be returned as Python strings, e.g. ``''`` values. This situation
319is not known to occur if the SQLAlchemy :class:`.LargeBinary` datatype is used
320consistently, however if a particular SQLite database has data that was
321inserted using the Pysqlite driver directly, or when using the SQLAlchemy
322:class:`.String` type which was later changed to :class:`.LargeBinary`, the
323table will not be consistently readable because SQLAlchemy's
324:class:`.LargeBinary` datatype does not handle strings so it has no way of
325"encoding" a value that is in string format.
327To deal with a SQLite table that has mixed string / binary data in the
328same column, use a custom type that will check each row individually::
330 from sqlalchemy import String
331 from sqlalchemy import TypeDecorator
334 class MixedBinary(TypeDecorator):
335 impl = String
336 cache_ok = True
338 def process_result_value(self, value, dialect):
339 if isinstance(value, str):
340 value = bytes(value, "utf-8")
341 elif value is not None:
342 value = bytes(value)
344 return value
346Then use the above ``MixedBinary`` datatype in the place where
347:class:`.LargeBinary` would normally be used.
349.. _pysqlite_serializable:
351Serializable isolation / Savepoints / Transactional DDL
352-------------------------------------------------------
354A newly revised version of this important section is now available
355at the top level of the SQLAlchemy SQLite documentation, in the section
356:ref:`sqlite_transactions`.
359.. _pysqlite_udfs:
361User-Defined Functions
362----------------------
364pysqlite supports a `create_function() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function>`_
365method that allows us to create our own user-defined functions (UDFs) in Python and use them directly in SQLite queries.
366These functions are registered with a specific DBAPI Connection.
368SQLAlchemy uses connection pooling with file-based SQLite databases, so we need to ensure that the UDF is attached to the
369connection when it is created. That is accomplished with an event listener::
371 from sqlalchemy import create_engine
372 from sqlalchemy import event
373 from sqlalchemy import text
376 def udf():
377 return "udf-ok"
380 engine = create_engine("sqlite:///./db_file")
383 @event.listens_for(engine, "connect")
384 def connect(conn, rec):
385 conn.create_function("udf", 0, udf)
388 for i in range(5):
389 with engine.connect() as conn:
390 print(conn.scalar(text("SELECT UDF()")))
392""" # noqa
394from __future__ import annotations
396import math
397import os
398import re
399from typing import Any
400from typing import Callable
401from typing import cast
402from typing import Optional
403from typing import Pattern
404from typing import TYPE_CHECKING
405from typing import TypeVar
406from typing import Union
408from .base import DATE
409from .base import DATETIME
410from .base import SQLiteDialect
411from ... import exc
412from ... import pool
413from ... import types as sqltypes
414from ... import util
415from ...util.typing import Self
417if TYPE_CHECKING:
418 from ...engine.interfaces import ConnectArgsType
419 from ...engine.interfaces import DBAPIConnection
420 from ...engine.interfaces import DBAPICursor
421 from ...engine.interfaces import DBAPIModule
422 from ...engine.interfaces import IsolationLevel
423 from ...engine.interfaces import VersionInfoType
424 from ...engine.url import URL
425 from ...pool.base import PoolProxiedConnection
426 from ...sql.type_api import _BindProcessorType
427 from ...sql.type_api import _ResultProcessorType
430class _SQLite_pysqliteTimeStamp(DATETIME):
431 def bind_processor( # type: ignore[override]
432 self, dialect: SQLiteDialect
433 ) -> Optional[_BindProcessorType[Any]]:
434 if dialect.native_datetime:
435 return None
436 else:
437 return DATETIME.bind_processor(self, dialect)
439 def result_processor( # type: ignore[override]
440 self, dialect: SQLiteDialect, coltype: object
441 ) -> Optional[_ResultProcessorType[Any]]:
442 if dialect.native_datetime:
443 return None
444 else:
445 return DATETIME.result_processor(self, dialect, coltype)
448class _SQLite_pysqliteDate(DATE):
449 def bind_processor( # type: ignore[override]
450 self, dialect: SQLiteDialect
451 ) -> Optional[_BindProcessorType[Any]]:
452 if dialect.native_datetime:
453 return None
454 else:
455 return DATE.bind_processor(self, dialect)
457 def result_processor( # type: ignore[override]
458 self, dialect: SQLiteDialect, coltype: object
459 ) -> Optional[_ResultProcessorType[Any]]:
460 if dialect.native_datetime:
461 return None
462 else:
463 return DATE.result_processor(self, dialect, coltype)
466class SQLiteDialect_pysqlite(SQLiteDialect):
467 default_paramstyle = "qmark"
468 supports_statement_cache = True
469 returns_native_bytes = True
471 colspecs = util.update_copy(
472 SQLiteDialect.colspecs,
473 {
474 sqltypes.Date: _SQLite_pysqliteDate,
475 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
476 },
477 )
479 description_encoding = None
481 driver = "pysqlite"
483 @classmethod
484 def import_dbapi(cls) -> DBAPIModule:
485 from sqlite3 import dbapi2 as sqlite
487 return cast("DBAPIModule", sqlite)
489 @classmethod
490 def _is_url_file_db(cls, url: URL) -> bool:
491 if (url.database and url.database != ":memory:") and (
492 url.query.get("mode", None) != "memory"
493 ):
494 return True
495 else:
496 return False
498 @classmethod
499 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
500 if cls._is_url_file_db(url):
501 return pool.QueuePool
502 else:
503 return pool.SingletonThreadPool
505 def _get_server_version_info(self, connection: Any) -> VersionInfoType:
506 return self.dbapi.sqlite_version_info # type: ignore
508 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
509 {
510 "AUTOCOMMIT": None,
511 }
512 )
514 def set_isolation_level(
515 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
516 ) -> None:
517 if level == "AUTOCOMMIT":
518 dbapi_connection.isolation_level = None
519 else:
520 dbapi_connection.isolation_level = ""
521 return super().set_isolation_level(dbapi_connection, level)
523 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
524 return dbapi_conn.isolation_level is None
526 def on_connect(self) -> Callable[[DBAPIConnection], None]:
527 def regexp(a: str, b: Optional[str]) -> Optional[bool]:
528 if b is None:
529 return None
530 return re.search(a, b) is not None
532 if self._get_server_version_info(None) >= (3, 9):
533 # sqlite must be greater than 3.8.3 for deterministic=True
534 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
535 # the check is more conservative since there were still issues
536 # with following 3.8 sqlite versions
537 create_func_kw = {"deterministic": True}
538 else:
539 create_func_kw = {}
541 def set_regexp(dbapi_connection: DBAPIConnection) -> None:
542 dbapi_connection.create_function(
543 "regexp", 2, regexp, **create_func_kw
544 )
546 def floor_func(dbapi_connection: DBAPIConnection) -> None:
547 # NOTE: floor is optionally present in sqlite 3.35+ , however
548 # as it is normally non-present we deliver floor() unconditionally
549 # for now.
550 # https://www.sqlite.org/lang_mathfunc.html
551 dbapi_connection.create_function(
552 "floor", 1, math.floor, **create_func_kw
553 )
555 fns = [set_regexp, floor_func]
557 def connect(conn: DBAPIConnection) -> None:
558 for fn in fns:
559 fn(conn)
561 return connect
563 def create_connect_args(self, url: URL) -> ConnectArgsType:
564 if url.username or url.password or url.host or url.port:
565 raise exc.ArgumentError(
566 "Invalid SQLite URL: %s\n"
567 "Valid SQLite URL forms are:\n"
568 " sqlite:///:memory: (or, sqlite://)\n"
569 " sqlite:///relative/path/to/file.db\n"
570 " sqlite:////absolute/path/to/file.db" % (url,)
571 )
573 # theoretically, this list can be augmented, at least as far as
574 # parameter names accepted by sqlite3/pysqlite, using
575 # inspect.getfullargspec(). for the moment this seems like overkill
576 # as these parameters don't change very often, and as always,
577 # parameters passed to connect_args will always go to the
578 # sqlite3/pysqlite driver.
579 pysqlite_args = [
580 ("uri", bool),
581 ("timeout", float),
582 ("isolation_level", str),
583 ("detect_types", int),
584 ("check_same_thread", bool),
585 ("cached_statements", int),
586 ]
587 opts = url.query
588 pysqlite_opts: dict[str, Any] = {}
589 for key, type_ in pysqlite_args:
590 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
592 if pysqlite_opts.get("uri", False):
593 uri_opts = dict(opts)
594 # here, we are actually separating the parameters that go to
595 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
596 # two names conflict? again, this seems to be not the case right
597 # now, and in the case that new names are added to
598 # either side which overlap, again the sqlite3/pysqlite parameters
599 # can be passed through connect_args instead of in the URL.
600 # If SQLite native URIs add a parameter like "timeout" that
601 # we already have listed here for the python driver, then we need
602 # to adjust for that here.
603 for key, type_ in pysqlite_args:
604 uri_opts.pop(key, None)
605 filename: str = url.database # type: ignore[assignment]
606 if uri_opts:
607 # sorting of keys is for unit test support
608 filename += "?" + (
609 "&".join(
610 "%s=%s" % (key, uri_opts[key])
611 for key in sorted(uri_opts)
612 )
613 )
614 else:
615 filename = url.database or ":memory:"
616 if filename != ":memory:":
617 filename = os.path.abspath(filename)
619 pysqlite_opts.setdefault(
620 "check_same_thread", not self._is_url_file_db(url)
621 )
623 return ([filename], pysqlite_opts)
625 def is_disconnect(
626 self,
627 e: DBAPIModule.Error,
628 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
629 cursor: Optional[DBAPICursor],
630 ) -> bool:
631 self.dbapi = cast("DBAPIModule", self.dbapi)
632 return isinstance(
633 e, self.dbapi.ProgrammingError
634 ) and "Cannot operate on a closed database." in str(e)
637dialect = SQLiteDialect_pysqlite
640class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
641 """numeric dialect for testing only
643 internal use only. This dialect is **NOT** supported by SQLAlchemy
644 and may change at any time.
646 """
648 supports_statement_cache = True
649 default_paramstyle = "numeric"
650 driver = "pysqlite_numeric"
652 _first_bind = ":1"
653 _not_in_statement_regexp: Optional[Pattern[str]] = None
655 def __init__(self, *arg: Any, **kw: Any) -> None:
656 kw.setdefault("paramstyle", "numeric")
657 super().__init__(*arg, **kw)
659 def create_connect_args(self, url: URL) -> ConnectArgsType:
660 arg, opts = super().create_connect_args(url)
661 opts["factory"] = self._fix_sqlite_issue_99953()
662 return arg, opts
664 def _fix_sqlite_issue_99953(self) -> Any:
665 import sqlite3
667 first_bind = self._first_bind
668 if self._not_in_statement_regexp:
669 nis = self._not_in_statement_regexp
671 def _test_sql(sql: str) -> None:
672 m = nis.search(sql)
673 assert not m, f"Found {nis.pattern!r} in {sql!r}"
675 else:
677 def _test_sql(sql: str) -> None:
678 pass
680 def _numeric_param_as_dict(
681 parameters: Any,
682 ) -> Union[dict[str, Any], tuple[Any, ...]]:
683 if parameters:
684 assert isinstance(parameters, tuple)
685 return {
686 str(idx): value for idx, value in enumerate(parameters, 1)
687 }
688 else:
689 return ()
691 class SQLiteFix99953Cursor(sqlite3.Cursor):
692 def execute(self, sql: str, parameters: Any = ()) -> Self:
693 _test_sql(sql)
694 if first_bind in sql:
695 parameters = _numeric_param_as_dict(parameters)
696 return super().execute(sql, parameters)
698 def executemany(self, sql: str, parameters: Any) -> Self:
699 _test_sql(sql)
700 if first_bind in sql:
701 parameters = [
702 _numeric_param_as_dict(p) for p in parameters
703 ]
704 return super().executemany(sql, parameters)
706 class SQLiteFix99953Connection(sqlite3.Connection):
707 _CursorT = TypeVar("_CursorT", bound=sqlite3.Cursor)
709 def cursor(
710 self,
711 factory: Optional[
712 Callable[[sqlite3.Connection], _CursorT]
713 ] = None,
714 ) -> _CursorT:
715 if factory is None:
716 factory = SQLiteFix99953Cursor # type: ignore[assignment]
717 return super().cursor(factory=factory) # type: ignore[return-value] # noqa[E501]
719 def execute(
720 self, sql: str, parameters: Any = ()
721 ) -> sqlite3.Cursor:
722 _test_sql(sql)
723 if first_bind in sql:
724 parameters = _numeric_param_as_dict(parameters)
725 return super().execute(sql, parameters)
727 def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor:
728 _test_sql(sql)
729 if first_bind in sql:
730 parameters = [
731 _numeric_param_as_dict(p) for p in parameters
732 ]
733 return super().executemany(sql, parameters)
735 return SQLiteFix99953Connection
738class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
739 """numeric dialect that uses $ for testing only
741 internal use only. This dialect is **NOT** supported by SQLAlchemy
742 and may change at any time.
744 """
746 supports_statement_cache = True
747 default_paramstyle = "numeric_dollar"
748 driver = "pysqlite_dollar"
750 _first_bind = "$1"
751 _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
753 def __init__(self, *arg: Any, **kw: Any) -> None:
754 kw.setdefault("paramstyle", "numeric_dollar")
755 super().__init__(*arg, **kw)