codekingpro/portable-devtools
115k
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
7
8
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
15
16 Note that ``pysqlite`` is the same driver as the ``sqlite3``
17 module included with the Python distribution.
18
19Driver
20------
21
22The ``sqlite3`` Python DBAPI is standard on all modern Python versions;
23for cPython and Pypy, no additional installation is necessary.
24
25
26Connect Strings
27---------------
28
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:
31
32.. sourcecode:: text
33
34 driver://user:pass@host/database
35
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::
39
40 # relative path
41 e = create_engine("sqlite:///path/to/database.db")
42
43An absolute path, which is denoted by starting with a slash, means you
44need **four** slashes::
45
46 # absolute path
47 e = create_engine("sqlite:////path/to/database.db")
48
49To use a Windows path, regular drive specifications and backslashes can be
50used. Double backslashes are probably needed::
51
52 # absolute path on Windows
53 e = create_engine("sqlite:///C:\\path\\to\\database.db")
54
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::
58
59 # in-memory database (note three slashes)
60 e = create_engine("sqlite:///:memory:")
61 # also in-memory database
62 e2 = create_engine("sqlite://")
63
64.. _pysqlite_uri_connections:
65
66URI Connections
67^^^^^^^^^^^^^^^
68
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)::
76
77 e = create_engine("sqlite:///file:path/to/database?mode=ro&uri=true")
78
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.
83
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::
92
93 e = create_engine(
94 "sqlite:///file:path/to/database?"
95 "check_same_thread=true&timeout=10&mode=ro&nolock=1&uri=true"
96 )
97
98Above, the pysqlite / sqlite3 DBAPI would be passed arguments as::
99
100 sqlite3.connect(
101 "file:path/to/database?mode=ro&nolock=1",
102 check_same_thread=True,
103 timeout=10,
104 uri=True,
105 )
106
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.
117
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.
123
124.. versionadded:: 1.3.9
125
126.. seealso::
127
128 `Uniform Resource Identifiers <https://www.sqlite.org/uri.html>`_ - in
129 the SQLite documentation
130
131.. _pysqlite_regexp:
132
133Regular Expression Support
134---------------------------
135
136.. versionadded:: 1.4
137
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.
142
143SQLAlchemy's implementation makes use of the pysqlite create_function_ hook
144as follows::
145
146
147 def regexp(a, b):
148 return re.search(a, b) is not None
149
150
151 sqlite_connection.create_function(
152 "regexp",
153 2,
154 regexp,
155 )
156
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.
161
162.. seealso::
163
164 `Python regular expressions`_: Documentation for Python's regular expression syntax.
165
166.. _create_function: https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
167
168.. _re.search: https://docs.python.org/3/library/re.html#re.search
169
170.. _Python regular expressions: https://docs.python.org/3/library/re.html#re.search
171
172
173
174Compatibility with sqlite3 "native" date and datetime types
175-----------------------------------------------------------
176
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.
190
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()::
194
195 engine = create_engine(
196 "sqlite://",
197 connect_args={
198 "detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES
199 },
200 native_datetime=True,
201 )
202
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.
209
210.. _pysqlite_threading_pooling:
211
212Threading/Pooling Behavior
213---------------------------
214
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.
219
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`.
226
227The SQLAlchemy ``pysqlite`` DBAPI establishes the connection pool differently
228based on the kind of SQLite database that's requested:
229
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.
239
240 .. versionchanged:: 2.0
241
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.
246
247Disabling Connection Pooling for File Databases
248^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
249
250Pooling may be disabled for a file based database by specifying the
251:class:`.NullPool` implementation for the :func:`_sa.create_engine.poolclass`
252parameter::
253
254 from sqlalchemy import NullPool
255
256 engine = create_engine("sqlite:///myfile.db", poolclass=NullPool)
257
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.
263
264Using a Memory Database in Multiple Threads
265^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
266
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``::
273
274 from sqlalchemy.pool import StaticPool
275
276 engine = create_engine(
277 "sqlite://",
278 connect_args={"check_same_thread": False},
279 poolclass=StaticPool,
280 )
281
282Note that using a ``:memory:`` database in multiple threads requires a recent
283version of SQLite.
284
285Using Temporary Tables with SQLite
286^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
287
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::
296
297 # maintain the same connection per thread
298 from sqlalchemy.pool import SingletonThreadPool
299
300 engine = create_engine("sqlite:///mydb.db", poolclass=SingletonThreadPool)
301
302
303 # maintain the same connection across all threads
304 from sqlalchemy.pool import StaticPool
305
306 engine = create_engine("sqlite:///mydb.db", poolclass=StaticPool)
307
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.
311
312
313Dealing with Mixed String / Binary Columns
314------------------------------------------------------
315
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.
328
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::
331
332 from sqlalchemy import String
333 from sqlalchemy import TypeDecorator
334
335
336 class MixedBinary(TypeDecorator):
337 impl = String
338 cache_ok = True
339
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)
345
346 return value
347
348Then use the above ``MixedBinary`` datatype in the place where
349:class:`.LargeBinary` would normally be used.
350
351.. _pysqlite_serializable:
352
353Serializable isolation / Savepoints / Transactional DDL
354-------------------------------------------------------
355
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`.
359
360
361.. _pysqlite_udfs:
362
363User-Defined Functions
364----------------------
365
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.
369
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::
372
373 from sqlalchemy import create_engine
374 from sqlalchemy import event
375 from sqlalchemy import text
376
377
378 def udf():
379 return "udf-ok"
380
381
382 engine = create_engine("sqlite:///./db_file")
383
384
385 @event.listens_for(engine, "connect")
386 def connect(conn, rec):
387 conn.create_function("udf", 0, udf)
388
389
390 for i in range(5):
391 with engine.connect() as conn:
392 print(conn.scalar(text("SELECT UDF()")))
393
394""" # noqa
395from __future__ import annotations
396
397import math
398import os
399import re
400from typing import Any
401from typing import Callable
402from typing import cast
403from typing import Optional
404from typing import Pattern
405from typing import TYPE_CHECKING
406from typing import TypeVar
407from typing import Union
408
409from .base import DATE
410from .base import DATETIME
411from .base import SQLiteDialect
412from ... import exc
413from ... import pool
414from ... import types as sqltypes
415from ... import util
416from ...util.typing import Self
417
418if TYPE_CHECKING:
419 from ...engine.interfaces import ConnectArgsType
420 from ...engine.interfaces import DBAPIConnection
421 from ...engine.interfaces import DBAPICursor
422 from ...engine.interfaces import DBAPIModule
423 from ...engine.interfaces import IsolationLevel
424 from ...engine.interfaces import VersionInfoType
425 from ...engine.url import URL
426 from ...pool.base import PoolProxiedConnection
427 from ...sql.type_api import _BindProcessorType
428 from ...sql.type_api import _ResultProcessorType
429
430
431class _SQLite_pysqliteTimeStamp(DATETIME):
432 def bind_processor( # type: ignore[override]
433 self, dialect: SQLiteDialect
434 ) -> Optional[_BindProcessorType[Any]]:
435 if dialect.native_datetime:
436 return None
437 else:
438 return DATETIME.bind_processor(self, dialect)
439
440 def result_processor( # type: ignore[override]
441 self, dialect: SQLiteDialect, coltype: object
442 ) -> Optional[_ResultProcessorType[Any]]:
443 if dialect.native_datetime:
444 return None
445 else:
446 return DATETIME.result_processor(self, dialect, coltype)
447
448
449class _SQLite_pysqliteDate(DATE):
450 def bind_processor( # type: ignore[override]
451 self, dialect: SQLiteDialect
452 ) -> Optional[_BindProcessorType[Any]]:
453 if dialect.native_datetime:
454 return None
455 else:
456 return DATE.bind_processor(self, dialect)
457
458 def result_processor( # type: ignore[override]
459 self, dialect: SQLiteDialect, coltype: object
460 ) -> Optional[_ResultProcessorType[Any]]:
461 if dialect.native_datetime:
462 return None
463 else:
464 return DATE.result_processor(self, dialect, coltype)
465
466
467class SQLiteDialect_pysqlite(SQLiteDialect):
468 default_paramstyle = "qmark"
469 supports_statement_cache = True
470 returns_native_bytes = True
471
472 colspecs = util.update_copy(
473 SQLiteDialect.colspecs,
474 {
475 sqltypes.Date: _SQLite_pysqliteDate,
476 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
477 },
478 )
479
480 description_encoding = None
481
482 driver = "pysqlite"
483
484 @classmethod
485 def import_dbapi(cls) -> DBAPIModule:
486 from sqlite3 import dbapi2 as sqlite
487
488 return cast("DBAPIModule", sqlite)
489
490 @classmethod
491 def _is_url_file_db(cls, url: URL) -> bool:
492 if (url.database and url.database != ":memory:") and (
493 url.query.get("mode", None) != "memory"
494 ):
495 return True
496 else:
497 return False
498
499 @classmethod
500 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
501 if cls._is_url_file_db(url):
502 return pool.QueuePool
503 else:
504 return pool.SingletonThreadPool
505
506 def _get_server_version_info(self, connection: Any) -> VersionInfoType:
507 return self.dbapi.sqlite_version_info # type: ignore
508
509 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
510 {
511 "AUTOCOMMIT": None, # type: ignore[dict-item]
512 }
513 )
514
515 def set_isolation_level(
516 self, dbapi_connection: DBAPIConnection, level: IsolationLevel
517 ) -> None:
518 if level == "AUTOCOMMIT":
519 dbapi_connection.isolation_level = None
520 else:
521 dbapi_connection.isolation_level = ""
522 return super().set_isolation_level(dbapi_connection, level)
523
524 def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool:
525 return dbapi_conn.isolation_level is None
526
527 def on_connect(self) -> Callable[[DBAPIConnection], None]:
528 def regexp(a: str, b: Optional[str]) -> Optional[bool]:
529 if b is None:
530 return None
531 return re.search(a, b) is not None
532
533 if util.py38 and self._get_server_version_info(None) >= (3, 9):
534 # sqlite must be greater than 3.8.3 for deterministic=True
535 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
536 # the check is more conservative since there were still issues
537 # with following 3.8 sqlite versions
538 create_func_kw = {"deterministic": True}
539 else:
540 create_func_kw = {}
541
542 def set_regexp(dbapi_connection: DBAPIConnection) -> None:
543 dbapi_connection.create_function(
544 "regexp", 2, regexp, **create_func_kw
545 )
546
547 def floor_func(dbapi_connection: DBAPIConnection) -> None:
548 # NOTE: floor is optionally present in sqlite 3.35+ , however
549 # as it is normally non-present we deliver floor() unconditionally
550 # for now.
551 # https://www.sqlite.org/lang_mathfunc.html
552 dbapi_connection.create_function(
553 "floor", 1, math.floor, **create_func_kw
554 )
555
556 fns = [set_regexp, floor_func]
557
558 def connect(conn: DBAPIConnection) -> None:
559 for fn in fns:
560 fn(conn)
561
562 return connect
563
564 def create_connect_args(self, url: URL) -> ConnectArgsType:
565 if url.username or url.password or url.host or url.port:
566 raise exc.ArgumentError(
567 "Invalid SQLite URL: %s\n"
568 "Valid SQLite URL forms are:\n"
569 " sqlite:///:memory: (or, sqlite://)\n"
570 " sqlite:///relative/path/to/file.db\n"
571 " sqlite:////absolute/path/to/file.db" % (url,)
572 )
573
574 # theoretically, this list can be augmented, at least as far as
575 # parameter names accepted by sqlite3/pysqlite, using
576 # inspect.getfullargspec(). for the moment this seems like overkill
577 # as these parameters don't change very often, and as always,
578 # parameters passed to connect_args will always go to the
579 # sqlite3/pysqlite driver.
580 pysqlite_args = [
581 ("uri", bool),
582 ("timeout", float),
583 ("isolation_level", str),
584 ("detect_types", int),
585 ("check_same_thread", bool),
586 ("cached_statements", int),
587 ]
588 opts = url.query
589 pysqlite_opts: dict[str, Any] = {}
590 for key, type_ in pysqlite_args:
591 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
592
593 if pysqlite_opts.get("uri", False):
594 uri_opts = dict(opts)
595 # here, we are actually separating the parameters that go to
596 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
597 # two names conflict? again, this seems to be not the case right
598 # now, and in the case that new names are added to
599 # either side which overlap, again the sqlite3/pysqlite parameters
600 # can be passed through connect_args instead of in the URL.
601 # If SQLite native URIs add a parameter like "timeout" that
602 # we already have listed here for the python driver, then we need
603 # to adjust for that here.
604 for key, type_ in pysqlite_args:
605 uri_opts.pop(key, None)
606 filename: str = url.database # type: ignore[assignment]
607 if uri_opts:
608 # sorting of keys is for unit test support
609 filename += "?" + (
610 "&".join(
611 "%s=%s" % (key, uri_opts[key])
612 for key in sorted(uri_opts)
613 )
614 )
615 else:
616 filename = url.database or ":memory:"
617 if filename != ":memory:":
618 filename = os.path.abspath(filename)
619
620 pysqlite_opts.setdefault(
621 "check_same_thread", not self._is_url_file_db(url)
622 )
623
624 return ([filename], pysqlite_opts)
625
626 def is_disconnect(
627 self,
628 e: DBAPIModule.Error,
629 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
630 cursor: Optional[DBAPICursor],
631 ) -> bool:
632 self.dbapi = cast("DBAPIModule", self.dbapi)
633 return isinstance(
634 e, self.dbapi.ProgrammingError
635 ) and "Cannot operate on a closed database." in str(e)
636
637
638dialect = SQLiteDialect_pysqlite
639
640
641class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
642 """numeric dialect for testing only
643
644 internal use only. This dialect is **NOT** supported by SQLAlchemy
645 and may change at any time.
646
647 """
648
649 supports_statement_cache = True
650 default_paramstyle = "numeric"
651 driver = "pysqlite_numeric"
652
653 _first_bind = ":1"
654 _not_in_statement_regexp: Optional[Pattern[str]] = None
655
656 def __init__(self, *arg: Any, **kw: Any) -> None:
657 kw.setdefault("paramstyle", "numeric")
658 super().__init__(*arg, **kw)
659
660 def create_connect_args(self, url: URL) -> ConnectArgsType:
661 arg, opts = super().create_connect_args(url)
662 opts["factory"] = self._fix_sqlite_issue_99953()
663 return arg, opts
664
665 def _fix_sqlite_issue_99953(self) -> Any:
666 import sqlite3
667
668 first_bind = self._first_bind
669 if self._not_in_statement_regexp:
670 nis = self._not_in_statement_regexp
671
672 def _test_sql(sql: str) -> None:
673 m = nis.search(sql)
674 assert not m, f"Found {nis.pattern!r} in {sql!r}"
675
676 else:
677
678 def _test_sql(sql: str) -> None:
679 pass
680
681 def _numeric_param_as_dict(
682 parameters: Any,
683 ) -> Union[dict[str, Any], tuple[Any, ...]]:
684 if parameters:
685 assert isinstance(parameters, tuple)
686 return {
687 str(idx): value for idx, value in enumerate(parameters, 1)
688 }
689 else:
690 return ()
691
692 class SQLiteFix99953Cursor(sqlite3.Cursor):
693 def execute(self, sql: str, parameters: Any = ()) -> Self:
694 _test_sql(sql)
695 if first_bind in sql:
696 parameters = _numeric_param_as_dict(parameters)
697 return super().execute(sql, parameters)
698
699 def executemany(self, sql: str, parameters: Any) -> Self:
700 _test_sql(sql)
701 if first_bind in sql:
702 parameters = [
703 _numeric_param_as_dict(p) for p in parameters
704 ]
705 return super().executemany(sql, parameters)
706
707 class SQLiteFix99953Connection(sqlite3.Connection):
708 _CursorT = TypeVar("_CursorT", bound=sqlite3.Cursor)
709
710 def cursor(
711 self,
712 factory: Optional[
713 Callable[[sqlite3.Connection], _CursorT]
714 ] = None,
715 ) -> _CursorT:
716 if factory is None:
717 factory = SQLiteFix99953Cursor # type: ignore[assignment]
718 return super().cursor(factory=factory) # type: ignore[return-value] # noqa[E501]
719
720 def execute(
721 self, sql: str, parameters: Any = ()
722 ) -> sqlite3.Cursor:
723 _test_sql(sql)
724 if first_bind in sql:
725 parameters = _numeric_param_as_dict(parameters)
726 return super().execute(sql, parameters)
727
728 def executemany(self, sql: str, parameters: Any) -> sqlite3.Cursor:
729 _test_sql(sql)
730 if first_bind in sql:
731 parameters = [
732 _numeric_param_as_dict(p) for p in parameters
733 ]
734 return super().executemany(sql, parameters)
735
736 return SQLiteFix99953Connection
737
738
739class _SQLiteDialect_pysqlite_dollar(_SQLiteDialect_pysqlite_numeric):
740 """numeric dialect that uses $ for testing only
741
742 internal use only. This dialect is **NOT** supported by SQLAlchemy
743 and may change at any time.
744
745 """
746
747 supports_statement_cache = True
748 default_paramstyle = "numeric_dollar"
749 driver = "pysqlite_dollar"
750
751 _first_bind = "$1"
752 _not_in_statement_regexp = re.compile(r"[^\d]:\d+")
753
754 def __init__(self, *arg: Any, **kw: Any) -> None:
755 kw.setdefault("paramstyle", "numeric_dollar")
756 super().__init__(*arg, **kw)
757