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