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-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
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 re-use 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
393from __future__ import annotations
395import math
396import os
397import re
398from typing import Any
399from typing import Callable
400from typing import cast
401from typing import Optional
402from typing import Pattern
403from typing import TYPE_CHECKING
404from typing import TypeVar
405from typing import Union
407from .base import DATE
408from .base import DATETIME
409from .base import SQLiteDialect
410from ... import exc
411from ... import pool
412from ... import types as sqltypes
413from ... import util
414from ...util.typing import Self
416if TYPE_CHECKING:
417 from ...engine.interfaces import ConnectArgsType
418 from ...engine.interfaces import DBAPIConnection
419 from ...engine.interfaces import DBAPICursor
420 from ...engine.interfaces import DBAPIModule
421 from ...engine.interfaces import IsolationLevel
422 from ...engine.interfaces import VersionInfoType
423 from ...engine.url import URL
424 from ...pool.base import PoolProxiedConnection
425 from ...sql.type_api import _BindProcessorType
426 from ...sql.type_api import _ResultProcessorType
429class _SQLite_pysqliteTimeStamp(DATETIME):
430 def bind_processor( # type: ignore[override]
431 self, dialect: SQLiteDialect
432 ) -> Optional[_BindProcessorType[Any]]:
433 if dialect.native_datetime:
434 return None
435 else:
436 return DATETIME.bind_processor(self, dialect)
438 def result_processor( # type: ignore[override]
439 self, dialect: SQLiteDialect, coltype: object
440 ) -> Optional[_ResultProcessorType[Any]]:
441 if dialect.native_datetime:
442 return None
443 else:
444 return DATETIME.result_processor(self, dialect, coltype)
447class _SQLite_pysqliteDate(DATE):
448 def bind_processor( # type: ignore[override]
449 self, dialect: SQLiteDialect
450 ) -> Optional[_BindProcessorType[Any]]:
451 if dialect.native_datetime:
452 return None
453 else:
454 return DATE.bind_processor(self, dialect)
456 def result_processor( # type: ignore[override]
457 self, dialect: SQLiteDialect, coltype: object
458 ) -> Optional[_ResultProcessorType[Any]]:
459 if dialect.native_datetime:
460 return None
461 else:
462 return DATE.result_processor(self, dialect, coltype)
465class SQLiteDialect_pysqlite(SQLiteDialect):
466 default_paramstyle = "qmark"
467 supports_statement_cache = True
468 returns_native_bytes = True
470 colspecs = util.update_copy(
471 SQLiteDialect.colspecs,
472 {
473 sqltypes.Date: _SQLite_pysqliteDate,
474 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
475 },
476 )
478 description_encoding = None
480 driver = "pysqlite"
482 @classmethod
483 def import_dbapi(cls) -> DBAPIModule:
484 from sqlite3 import dbapi2 as sqlite
486 return cast("DBAPIModule", sqlite)
488 @classmethod
489 def _is_url_file_db(cls, url: URL) -> bool:
490 if (url.database and url.database != ":memory:") and (
491 url.query.get("mode", None) != "memory"
492 ):
493 return True
494 else:
495 return False
497 @classmethod
498 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
499 if cls._is_url_file_db(url):
500 return pool.QueuePool
501 else:
502 return pool.SingletonThreadPool
504 def _get_server_version_info(self, connection: Any) -> VersionInfoType:
505 return self.dbapi.sqlite_version_info # type: ignore
507 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
508 {
509 "AUTOCOMMIT": None,
510 }
511 )
513 def set_isolation_level(
514 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
515 ) -> None:
516 if level == "AUTOCOMMIT":
517 dbapi_connection.isolation_level = None
518 else:
519 dbapi_connection.isolation_level = ""
520 return super().set_isolation_level(dbapi_connection, level)
522 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
523 return dbapi_conn.isolation_level is None
525 def on_connect(self) -> Callable[[DBAPIConnection], None]:
526 def regexp(a: str, b: Optional[str]) -> Optional[bool]:
527 if b is None:
528 return None
529 return re.search(a, b) is not None
531 if self._get_server_version_info(None) >= (3, 9):
532 # sqlite must be greater than 3.8.3 for deterministic=True
533 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
534 # the check is more conservative since there were still issues
535 # with following 3.8 sqlite versions
536 create_func_kw = {"deterministic": True}
537 else:
538 create_func_kw = {}
540 def set_regexp(dbapi_connection: DBAPIConnection) -> None:
541 dbapi_connection.create_function(
542 "regexp", 2, regexp, **create_func_kw
543 )
545 def floor_func(dbapi_connection: DBAPIConnection) -> None:
546 # NOTE: floor is optionally present in sqlite 3.35+ , however
547 # as it is normally non-present we deliver floor() unconditionally
548 # for now.
549 # https://www.sqlite.org/lang_mathfunc.html
550 dbapi_connection.create_function(
551 "floor", 1, math.floor, **create_func_kw
552 )
554 fns = [set_regexp, floor_func]
556 def connect(conn: DBAPIConnection) -> None:
557 for fn in fns:
558 fn(conn)
560 return connect
562 def create_connect_args(self, url: URL) -> ConnectArgsType:
563 if url.username or url.password or url.host or url.port:
564 raise exc.ArgumentError(
565 "Invalid SQLite URL: %s\n"
566 "Valid SQLite URL forms are:\n"
567 " sqlite:///:memory: (or, sqlite://)\n"
568 " sqlite:///relative/path/to/file.db\n"
569 " sqlite:////absolute/path/to/file.db" % (url,)
570 )
572 # theoretically, this list can be augmented, at least as far as
573 # parameter names accepted by sqlite3/pysqlite, using
574 # inspect.getfullargspec(). for the moment this seems like overkill
575 # as these parameters don't change very often, and as always,
576 # parameters passed to connect_args will always go to the
577 # sqlite3/pysqlite driver.
578 pysqlite_args = [
579 ("uri", bool),
580 ("timeout", float),
581 ("isolation_level", str),
582 ("detect_types", int),
583 ("check_same_thread", bool),
584 ("cached_statements", int),
585 ]
586 opts = url.query
587 pysqlite_opts: dict[str, Any] = {}
588 for key, type_ in pysqlite_args:
589 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
591 if pysqlite_opts.get("uri", False):
592 uri_opts = dict(opts)
593 # here, we are actually separating the parameters that go to
594 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
595 # two names conflict? again, this seems to be not the case right
596 # now, and in the case that new names are added to
597 # either side which overlap, again the sqlite3/pysqlite parameters
598 # can be passed through connect_args instead of in the URL.
599 # If SQLite native URIs add a parameter like "timeout" that
600 # we already have listed here for the python driver, then we need
601 # to adjust for that here.
602 for key, type_ in pysqlite_args:
603 uri_opts.pop(key, None)
604 filename: str = url.database # type: ignore[assignment]
605 if uri_opts:
606 # sorting of keys is for unit test support
607 filename += "?" + (
608 "&".join(
609 "%s=%s" % (key, uri_opts[key])
610 for key in sorted(uri_opts)
611 )
612 )
613 else:
614 filename = url.database or ":memory:"
615 if filename != ":memory:":
616 filename = os.path.abspath(filename)
618 pysqlite_opts.setdefault(
619 "check_same_thread", not self._is_url_file_db(url)
620 )
622 return ([filename], pysqlite_opts)
624 def is_disconnect(
625 self,
626 e: DBAPIModule.Error,
627 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
628 cursor: Optional[DBAPICursor],
629 ) -> bool:
630 self.dbapi = cast("DBAPIModule", self.dbapi)
631 return isinstance(
632 e, self.dbapi.ProgrammingError
633 ) and "Cannot operate on a closed database." in str(e)
636dialect = SQLiteDialect_pysqlite
639class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
640 """numeric dialect for testing only
642 internal use only. This dialect is **NOT** supported by SQLAlchemy
643 and may change at any time.
645 """
647 supports_statement_cache = True
648 default_paramstyle = "numeric"
649 driver = "pysqlite_numeric"
651 _first_bind = ":1"
652 _not_in_statement_regexp: Optional[Pattern[str]] = None
654 def __init__(self, *arg: Any, **kw: Any) -> None:
655 kw.setdefault("paramstyle", "numeric")
656 super().__init__(*arg, **kw)
658 def create_connect_args(self, url: URL) -> ConnectArgsType:
659 arg, opts = super().create_connect_args(url)
660 opts["factory"] = self._fix_sqlite_issue_99953()
661 return arg, opts
663 def _fix_sqlite_issue_99953(self) -> Any:
664 import sqlite3
666 first_bind = self._first_bind
667 if self._not_in_statement_regexp:
668 nis = self._not_in_statement_regexp
670 def _test_sql(sql: str) -> None:
671 m = nis.search(sql)
672 assert not m, f"Found {nis.pattern!r} in {sql!r}"
674 else:
676 def _test_sql(sql: str) -> None:
677 pass
679 def _numeric_param_as_dict(
680 parameters: Any,
681 ) -> Union[dict[str, Any], tuple[Any, ...]]:
682 if parameters:
683 assert isinstance(parameters, tuple)
684 return {
685 str(idx): value for idx, value in enumerate(parameters, 1)
686 }
687 else:
688 return ()
690 class SQLiteFix99953Cursor(sqlite3.Cursor):
691 def execute(self, sql: str, parameters: Any = ()) -> Self:
692 _test_sql(sql)
693 if first_bind in sql:
694 parameters = _numeric_param_as_dict(parameters)
695 return super().execute(sql, parameters)
697 def executemany(self, sql: str, parameters: Any) -> Self:
698 _test_sql(sql)
699 if first_bind in sql:
700 parameters = [
701 _numeric_param_as_dict(p) for p in parameters
702 ]
703 return super().executemany(sql, parameters)
705 class SQLiteFix99953Connection(sqlite3.Connection):
706 _CursorT = TypeVar("_CursorT", bound=sqlite3.Cursor)
708 def cursor(
709 self,
710 factory: Optional[
711 Callable[[sqlite3.Connection], _CursorT]
712 ] = None,
713 ) -> _CursorT:
714 if factory is None:
715 factory = SQLiteFix99953Cursor # type: ignore[assignment]
716 return super().cursor(factory=factory) # type: ignore[return-value] # noqa[E501]
718 def execute(
719 self, sql: str, parameters: Any = ()
720 ) -> sqlite3.Cursor:
721 _test_sql(sql)
722 if first_bind in sql:
723 parameters = _numeric_param_as_dict(parameters)
724 return super().execute(sql, parameters)
726 def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor:
727 _test_sql(sql)
728 if first_bind in sql:
729 parameters = [
730 _numeric_param_as_dict(p) for p in parameters
731 ]
732 return super().executemany(sql, parameters)
734 return SQLiteFix99953Connection
737class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
738 """numeric dialect that uses $ for testing only
740 internal use only. This dialect is **NOT** supported by SQLAlchemy
741 and may change at any time.
743 """
745 supports_statement_cache = True
746 default_paramstyle = "numeric_dollar"
747 driver = "pysqlite_dollar"
749 _first_bind = "$1"
750 _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
752 def __init__(self, *arg: Any, **kw: Any) -> None:
753 kw.setdefault("paramstyle", "numeric_dollar")
754 super().__init__(*arg, **kw)