codekingpro/portable-devtools
114k
1# dialects/sqlite/pysqlite.py
2# Copyright (C) 2005-2024 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# mypy: ignore-errors
8
9
10r"""
11.. dialect:: sqlite+pysqlite
12 :name: pysqlite
13 :dbapi: sqlite3
14 :connectstring: sqlite+pysqlite:///file_path
15 :url: https://docs.python.org/library/sqlite3.html
16
17 Note that ``pysqlite`` is the same driver as the ``sqlite3``
18 module included with the Python distribution.
19
20Driver
21------
22
23The ``sqlite3`` Python DBAPI is standard on all modern Python versions;
24for cPython and Pypy, no additional installation is necessary.
25
26
27Connect Strings
28---------------
29
30The file specification for the SQLite database is taken as the "database"
31portion of the URL. Note that the format of a SQLAlchemy url is::
32
33 driver://user:pass@host/database
34
35This means that the actual filename to be used starts with the characters to
36the **right** of the third slash. So connecting to a relative filepath
37looks like::
38
39 # relative path
40 e = create_engine('sqlite:///path/to/database.db')
41
42An absolute path, which is denoted by starting with a slash, means you
43need **four** slashes::
44
45 # absolute path
46 e = create_engine('sqlite:////path/to/database.db')
47
48To use a Windows path, regular drive specifications and backslashes can be
49used. Double backslashes are probably needed::
50
51 # absolute path on Windows
52 e = create_engine('sqlite:///C:\\path\\to\\database.db')
53
54To use sqlite ``:memory:`` database specify it as the filename using
55``sqlite://:memory:``. It's also the default if no filepath is
56present, specifying only ``sqlite://`` and nothing else::
57
58 # in-memory database
59 e = create_engine('sqlite://:memory:')
60 # also in-memory database
61 e2 = create_engine('sqlite://')
62
63.. _pysqlite_uri_connections:
64
65URI Connections
66^^^^^^^^^^^^^^^
67
68Modern versions of SQLite support an alternative system of connecting using a
69`driver level URI <https://www.sqlite.org/uri.html>`_, which has the advantage
70that additional driver-level arguments can be passed including options such as
71"read only". The Python sqlite3 driver supports this mode under modern Python
723 versions. The SQLAlchemy pysqlite driver supports this mode of use by
73specifying "uri=true" in the URL query string. The SQLite-level "URI" is kept
74as the "database" portion of the SQLAlchemy url (that is, following a slash)::
75
76 e = create_engine("sqlite:///file:path/to/database?mode=ro&uri=true")
77
78.. note:: The "uri=true" parameter must appear in the **query string**
79 of the URL. It will not currently work as expected if it is only
80 present in the :paramref:`_sa.create_engine.connect_args`
81 parameter dictionary.
82
83The logic reconciles the simultaneous presence of SQLAlchemy's query string and
84SQLite's query string by separating out the parameters that belong to the
85Python sqlite3 driver vs. those that belong to the SQLite URI. This is
86achieved through the use of a fixed list of parameters known to be accepted by
87the Python side of the driver. For example, to include a URL that indicates
88the Python sqlite3 "timeout" and "check_same_thread" parameters, along with the
89SQLite "mode" and "nolock" parameters, they can all be passed together on the
90query string::
91
92 e = create_engine(
93 "sqlite:///file:path/to/database?"
94 "check_same_thread=true&timeout=10&mode=ro&nolock=1&uri=true"
95 )
96
97Above, the pysqlite / sqlite3 DBAPI would be passed arguments as::
98
99 sqlite3.connect(
100 "file:path/to/database?mode=ro&nolock=1",
101 check_same_thread=True, timeout=10, uri=True
102 )
103
104Regarding future parameters added to either the Python or native drivers. new
105parameter names added to the SQLite URI scheme should be automatically
106accommodated by this scheme. New parameter names added to the Python driver
107side can be accommodated by specifying them in the
108:paramref:`_sa.create_engine.connect_args` dictionary,
109until dialect support is
110added by SQLAlchemy. For the less likely case that the native SQLite driver
111adds a new parameter name that overlaps with one of the existing, known Python
112driver parameters (such as "timeout" perhaps), SQLAlchemy's dialect would
113require adjustment for the URL scheme to continue to support this.
114
115As is always the case for all SQLAlchemy dialects, the entire "URL" process
116can be bypassed in :func:`_sa.create_engine` through the use of the
117:paramref:`_sa.create_engine.creator`
118parameter which allows for a custom callable
119that creates a Python sqlite3 driver level connection directly.
120
121.. versionadded:: 1.3.9
122
123.. seealso::
124
125 `Uniform Resource Identifiers <https://www.sqlite.org/uri.html>`_ - in
126 the SQLite documentation
127
128.. _pysqlite_regexp:
129
130Regular Expression Support
131---------------------------
132
133.. versionadded:: 1.4
134
135Support for the :meth:`_sql.ColumnOperators.regexp_match` operator is provided
136using Python's re.search_ function. SQLite itself does not include a working
137regular expression operator; instead, it includes a non-implemented placeholder
138operator ``REGEXP`` that calls a user-defined function that must be provided.
139
140SQLAlchemy's implementation makes use of the pysqlite create_function_ hook
141as follows::
142
143
144 def regexp(a, b):
145 return re.search(a, b) is not None
146
147 sqlite_connection.create_function(
148 "regexp", 2, regexp,
149 )
150
151There is currently no support for regular expression flags as a separate
152argument, as these are not supported by SQLite's REGEXP operator, however these
153may be included inline within the regular expression string. See `Python regular expressions`_ for
154details.
155
156.. seealso::
157
158 `Python regular expressions`_: Documentation for Python's regular expression syntax.
159
160.. _create_function: https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
161
162.. _re.search: https://docs.python.org/3/library/re.html#re.search
163
164.. _Python regular expressions: https://docs.python.org/3/library/re.html#re.search
165
166
167
168Compatibility with sqlite3 "native" date and datetime types
169-----------------------------------------------------------
170
171The pysqlite driver includes the sqlite3.PARSE_DECLTYPES and
172sqlite3.PARSE_COLNAMES options, which have the effect of any column
173or expression explicitly cast as "date" or "timestamp" will be converted
174to a Python date or datetime object. The date and datetime types provided
175with the pysqlite dialect are not currently compatible with these options,
176since they render the ISO date/datetime including microseconds, which
177pysqlite's driver does not. Additionally, SQLAlchemy does not at
178this time automatically render the "cast" syntax required for the
179freestanding functions "current_timestamp" and "current_date" to return
180datetime/date types natively. Unfortunately, pysqlite
181does not provide the standard DBAPI types in ``cursor.description``,
182leaving SQLAlchemy with no way to detect these types on the fly
183without expensive per-row type checks.
184
185Keeping in mind that pysqlite's parsing option is not recommended,
186nor should be necessary, for use with SQLAlchemy, usage of PARSE_DECLTYPES
187can be forced if one configures "native_datetime=True" on create_engine()::
188
189 engine = create_engine('sqlite://',
190 connect_args={'detect_types':
191 sqlite3.PARSE_DECLTYPES|sqlite3.PARSE_COLNAMES},
192 native_datetime=True
193 )
194
195With this flag enabled, the DATE and TIMESTAMP types (but note - not the
196DATETIME or TIME types...confused yet ?) will not perform any bind parameter
197or result processing. Execution of "func.current_date()" will return a string.
198"func.current_timestamp()" is registered as returning a DATETIME type in
199SQLAlchemy, so this function still receives SQLAlchemy-level result
200processing.
201
202.. _pysqlite_threading_pooling:
203
204Threading/Pooling Behavior
205---------------------------
206
207The ``sqlite3`` DBAPI by default prohibits the use of a particular connection
208in a thread which is not the one in which it was created. As SQLite has
209matured, it's behavior under multiple threads has improved, and even includes
210options for memory only databases to be used in multiple threads.
211
212The thread prohibition is known as "check same thread" and may be controlled
213using the ``sqlite3`` parameter ``check_same_thread``, which will disable or
214enable this check. SQLAlchemy's default behavior here is to set
215``check_same_thread`` to ``False`` automatically whenever a file-based database
216is in use, to establish compatibility with the default pool class
217:class:`.QueuePool`.
218
219The SQLAlchemy ``pysqlite`` DBAPI establishes the connection pool differently
220based on the kind of SQLite database that's requested:
221
222* When a ``:memory:`` SQLite database is specified, the dialect by default
223 will use :class:`.SingletonThreadPool`. This pool maintains a single
224 connection per thread, so that all access to the engine within the current
225 thread use the same ``:memory:`` database - other threads would access a
226 different ``:memory:`` database. The ``check_same_thread`` parameter
227 defaults to ``True``.
228* When a file-based database is specified, the dialect will use
229 :class:`.QueuePool` as the source of connections. at the same time,
230 the ``check_same_thread`` flag is set to False by default unless overridden.
231
232 .. versionchanged:: 2.0
233
234 SQLite file database engines now use :class:`.QueuePool` by default.
235 Previously, :class:`.NullPool` were used. The :class:`.NullPool` class
236 may be used by specifying it via the
237 :paramref:`_sa.create_engine.poolclass` parameter.
238
239Disabling Connection Pooling for File Databases
240^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
241
242Pooling may be disabled for a file based database by specifying the
243:class:`.NullPool` implementation for the :func:`_sa.create_engine.poolclass`
244parameter::
245
246 from sqlalchemy import NullPool
247 engine = create_engine("sqlite:///myfile.db", poolclass=NullPool)
248
249It's been observed that the :class:`.NullPool` implementation incurs an
250extremely small performance overhead for repeated checkouts due to the lack of
251connection re-use implemented by :class:`.QueuePool`. However, it still
252may be beneficial to use this class if the application is experiencing
253issues with files being locked.
254
255Using a Memory Database in Multiple Threads
256^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
257
258To use a ``:memory:`` database in a multithreaded scenario, the same
259connection object must be shared among threads, since the database exists
260only within the scope of that connection. The
261:class:`.StaticPool` implementation will maintain a single connection
262globally, and the ``check_same_thread`` flag can be passed to Pysqlite
263as ``False``::
264
265 from sqlalchemy.pool import StaticPool
266 engine = create_engine('sqlite://',
267 connect_args={'check_same_thread':False},
268 poolclass=StaticPool)
269
270Note that using a ``:memory:`` database in multiple threads requires a recent
271version of SQLite.
272
273Using Temporary Tables with SQLite
274^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
275
276Due to the way SQLite deals with temporary tables, if you wish to use a
277temporary table in a file-based SQLite database across multiple checkouts
278from the connection pool, such as when using an ORM :class:`.Session` where
279the temporary table should continue to remain after :meth:`.Session.commit` or
280:meth:`.Session.rollback` is called, a pool which maintains a single
281connection must be used. Use :class:`.SingletonThreadPool` if the scope is
282only needed within the current thread, or :class:`.StaticPool` is scope is
283needed within multiple threads for this case::
284
285 # maintain the same connection per thread
286 from sqlalchemy.pool import SingletonThreadPool
287 engine = create_engine('sqlite:///mydb.db',
288 poolclass=SingletonThreadPool)
289
290
291 # maintain the same connection across all threads
292 from sqlalchemy.pool import StaticPool
293 engine = create_engine('sqlite:///mydb.db',
294 poolclass=StaticPool)
295
296Note that :class:`.SingletonThreadPool` should be configured for the number
297of threads that are to be used; beyond that number, connections will be
298closed out in a non deterministic way.
299
300
301Dealing with Mixed String / Binary Columns
302------------------------------------------------------
303
304The SQLite database is weakly typed, and as such it is possible when using
305binary values, which in Python are represented as ``b'some string'``, that a
306particular SQLite database can have data values within different rows where
307some of them will be returned as a ``b''`` value by the Pysqlite driver, and
308others will be returned as Python strings, e.g. ``''`` values. This situation
309is not known to occur if the SQLAlchemy :class:`.LargeBinary` datatype is used
310consistently, however if a particular SQLite database has data that was
311inserted using the Pysqlite driver directly, or when using the SQLAlchemy
312:class:`.String` type which was later changed to :class:`.LargeBinary`, the
313table will not be consistently readable because SQLAlchemy's
314:class:`.LargeBinary` datatype does not handle strings so it has no way of
315"encoding" a value that is in string format.
316
317To deal with a SQLite table that has mixed string / binary data in the
318same column, use a custom type that will check each row individually::
319
320 from sqlalchemy import String
321 from sqlalchemy import TypeDecorator
322
323 class MixedBinary(TypeDecorator):
324 impl = String
325 cache_ok = True
326
327 def process_result_value(self, value, dialect):
328 if isinstance(value, str):
329 value = bytes(value, 'utf-8')
330 elif value is not None:
331 value = bytes(value)
332
333 return value
334
335Then use the above ``MixedBinary`` datatype in the place where
336:class:`.LargeBinary` would normally be used.
337
338.. _pysqlite_serializable:
339
340Serializable isolation / Savepoints / Transactional DDL
341-------------------------------------------------------
342
343In the section :ref:`sqlite_concurrency`, we refer to the pysqlite
344driver's assortment of issues that prevent several features of SQLite
345from working correctly. The pysqlite DBAPI driver has several
346long-standing bugs which impact the correctness of its transactional
347behavior. In its default mode of operation, SQLite features such as
348SERIALIZABLE isolation, transactional DDL, and SAVEPOINT support are
349non-functional, and in order to use these features, workarounds must
350be taken.
351
352The issue is essentially that the driver attempts to second-guess the user's
353intent, failing to start transactions and sometimes ending them prematurely, in
354an effort to minimize the SQLite databases's file locking behavior, even
355though SQLite itself uses "shared" locks for read-only activities.
356
357SQLAlchemy chooses to not alter this behavior by default, as it is the
358long-expected behavior of the pysqlite driver; if and when the pysqlite
359driver attempts to repair these issues, that will be more of a driver towards
360defaults for SQLAlchemy.
361
362The good news is that with a few events, we can implement transactional
363support fully, by disabling pysqlite's feature entirely and emitting BEGIN
364ourselves. This is achieved using two event listeners::
365
366 from sqlalchemy import create_engine, event
367
368 engine = create_engine("sqlite:///myfile.db")
369
370 @event.listens_for(engine, "connect")
371 def do_connect(dbapi_connection, connection_record):
372 # disable pysqlite's emitting of the BEGIN statement entirely.
373 # also stops it from emitting COMMIT before any DDL.
374 dbapi_connection.isolation_level = None
375
376 @event.listens_for(engine, "begin")
377 def do_begin(conn):
378 # emit our own BEGIN
379 conn.exec_driver_sql("BEGIN")
380
381.. warning:: When using the above recipe, it is advised to not use the
382 :paramref:`.Connection.execution_options.isolation_level` setting on
383 :class:`_engine.Connection` and :func:`_sa.create_engine`
384 with the SQLite driver,
385 as this function necessarily will also alter the ".isolation_level" setting.
386
387
388Above, we intercept a new pysqlite connection and disable any transactional
389integration. Then, at the point at which SQLAlchemy knows that transaction
390scope is to begin, we emit ``"BEGIN"`` ourselves.
391
392When we take control of ``"BEGIN"``, we can also control directly SQLite's
393locking modes, introduced at
394`BEGIN TRANSACTION <https://sqlite.org/lang_transaction.html>`_,
395by adding the desired locking mode to our ``"BEGIN"``::
396
397 @event.listens_for(engine, "begin")
398 def do_begin(conn):
399 conn.exec_driver_sql("BEGIN EXCLUSIVE")
400
401.. seealso::
402
403 `BEGIN TRANSACTION <https://sqlite.org/lang_transaction.html>`_ -
404 on the SQLite site
405
406 `sqlite3 SELECT does not BEGIN a transaction <https://bugs.python.org/issue9924>`_ -
407 on the Python bug tracker
408
409 `sqlite3 module breaks transactions and potentially corrupts data <https://bugs.python.org/issue10740>`_ -
410 on the Python bug tracker
411
412.. _pysqlite_udfs:
413
414User-Defined Functions
415----------------------
416
417pysqlite supports a `create_function() <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function>`_
418method that allows us to create our own user-defined functions (UDFs) in Python and use them directly in SQLite queries.
419These functions are registered with a specific DBAPI Connection.
420
421SQLAlchemy uses connection pooling with file-based SQLite databases, so we need to ensure that the UDF is attached to the
422connection when it is created. That is accomplished with an event listener::
423
424 from sqlalchemy import create_engine
425 from sqlalchemy import event
426 from sqlalchemy import text
427
428
429 def udf():
430 return "udf-ok"
431
432
433 engine = create_engine("sqlite:///./db_file")
434
435
436 @event.listens_for(engine, "connect")
437 def connect(conn, rec):
438 conn.create_function("udf", 0, udf)
439
440
441 for i in range(5):
442 with engine.connect() as conn:
443 print(conn.scalar(text("SELECT UDF()")))
444
445
446""" # noqa
447
448import math
449import os
450import re
451
452from .base import DATE
453from .base import DATETIME
454from .base import SQLiteDialect
455from ... import exc
456from ... import pool
457from ... import types as sqltypes
458from ... import util
459
460
461class _SQLite_pysqliteTimeStamp(DATETIME):
462 def bind_processor(self, dialect):
463 if dialect.native_datetime:
464 return None
465 else:
466 return DATETIME.bind_processor(self, dialect)
467
468 def result_processor(self, dialect, coltype):
469 if dialect.native_datetime:
470 return None
471 else:
472 return DATETIME.result_processor(self, dialect, coltype)
473
474
475class _SQLite_pysqliteDate(DATE):
476 def bind_processor(self, dialect):
477 if dialect.native_datetime:
478 return None
479 else:
480 return DATE.bind_processor(self, dialect)
481
482 def result_processor(self, dialect, coltype):
483 if dialect.native_datetime:
484 return None
485 else:
486 return DATE.result_processor(self, dialect, coltype)
487
488
489class SQLiteDialect_pysqlite(SQLiteDialect):
490 default_paramstyle = "qmark"
491 supports_statement_cache = True
492 returns_native_bytes = True
493
494 colspecs = util.update_copy(
495 SQLiteDialect.colspecs,
496 {
497 sqltypes.Date: _SQLite_pysqliteDate,
498 sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp,
499 },
500 )
501
502 description_encoding = None
503
504 driver = "pysqlite"
505
506 @classmethod
507 def import_dbapi(cls):
508 from sqlite3 import dbapi2 as sqlite
509
510 return sqlite
511
512 @classmethod
513 def _is_url_file_db(cls, url):
514 if (url.database and url.database != ":memory:") and (
515 url.query.get("mode", None) != "memory"
516 ):
517 return True
518 else:
519 return False
520
521 @classmethod
522 def get_pool_class(cls, url):
523 if cls._is_url_file_db(url):
524 return pool.QueuePool
525 else:
526 return pool.SingletonThreadPool
527
528 def _get_server_version_info(self, connection):
529 return self.dbapi.sqlite_version_info
530
531 _isolation_lookup = SQLiteDialect._isolation_lookup.union(
532 {
533 "AUTOCOMMIT": None,
534 }
535 )
536
537 def set_isolation_level(self, dbapi_connection, level):
538 if level == "AUTOCOMMIT":
539 dbapi_connection.isolation_level = None
540 else:
541 dbapi_connection.isolation_level = ""
542 return super().set_isolation_level(dbapi_connection, level)
543
544 def on_connect(self):
545 def regexp(a, b):
546 if b is None:
547 return None
548 return re.search(a, b) is not None
549
550 if util.py38 and self._get_server_version_info(None) >= (3, 9):
551 # sqlite must be greater than 3.8.3 for deterministic=True
552 # https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function
553 # the check is more conservative since there were still issues
554 # with following 3.8 sqlite versions
555 create_func_kw = {"deterministic": True}
556 else:
557 create_func_kw = {}
558
559 def set_regexp(dbapi_connection):
560 dbapi_connection.create_function(
561 "regexp", 2, regexp, **create_func_kw
562 )
563
564 def floor_func(dbapi_connection):
565 # NOTE: floor is optionally present in sqlite 3.35+ , however
566 # as it is normally non-present we deliver floor() unconditionally
567 # for now.
568 # https://www.sqlite.org/lang_mathfunc.html
569 dbapi_connection.create_function(
570 "floor", 1, math.floor, **create_func_kw
571 )
572
573 fns = [set_regexp, floor_func]
574
575 def connect(conn):
576 for fn in fns:
577 fn(conn)
578
579 return connect
580
581 def create_connect_args(self, url):
582 if url.username or url.password or url.host or url.port:
583 raise exc.ArgumentError(
584 "Invalid SQLite URL: %s\n"
585 "Valid SQLite URL forms are:\n"
586 " sqlite:///:memory: (or, sqlite://)\n"
587 " sqlite:///relative/path/to/file.db\n"
588 " sqlite:////absolute/path/to/file.db" % (url,)
589 )
590
591 # theoretically, this list can be augmented, at least as far as
592 # parameter names accepted by sqlite3/pysqlite, using
593 # inspect.getfullargspec(). for the moment this seems like overkill
594 # as these parameters don't change very often, and as always,
595 # parameters passed to connect_args will always go to the
596 # sqlite3/pysqlite driver.
597 pysqlite_args = [
598 ("uri", bool),
599 ("timeout", float),
600 ("isolation_level", str),
601 ("detect_types", int),
602 ("check_same_thread", bool),
603 ("cached_statements", int),
604 ]
605 opts = url.query
606 pysqlite_opts = {}
607 for key, type_ in pysqlite_args:
608 util.coerce_kw_type(opts, key, type_, dest=pysqlite_opts)
609
610 if pysqlite_opts.get("uri", False):
611 uri_opts = dict(opts)
612 # here, we are actually separating the parameters that go to
613 # sqlite3/pysqlite vs. those that go the SQLite URI. What if
614 # two names conflict? again, this seems to be not the case right
615 # now, and in the case that new names are added to
616 # either side which overlap, again the sqlite3/pysqlite parameters
617 # can be passed through connect_args instead of in the URL.
618 # If SQLite native URIs add a parameter like "timeout" that
619 # we already have listed here for the python driver, then we need
620 # to adjust for that here.
621 for key, type_ in pysqlite_args:
622 uri_opts.pop(key, None)
623 filename = url.database
624 if uri_opts:
625 # sorting of keys is for unit test support
626 filename += "?" + (
627 "&".join(
628 "%s=%s" % (key, uri_opts[key])
629 for key in sorted(uri_opts)
630 )
631 )
632 else:
633 filename = url.database or ":memory:"
634 if filename != ":memory:":
635 filename = os.path.abspath(filename)
636
637 pysqlite_opts.setdefault(
638 "check_same_thread", not self._is_url_file_db(url)
639 )
640
641 return ([filename], pysqlite_opts)
642
643 def is_disconnect(self, e, connection, cursor):
644 return isinstance(
645 e, self.dbapi.ProgrammingError
646 ) and "Cannot operate on a closed database." in str(e)
647
648
649dialect = SQLiteDialect_pysqlite
650
651
652class _SQLiteDialect_pysqlite_numeric(SQLiteDialect_pysqlite):
653 """numeric dialect for testing only
654
655 internal use only. This dialect is **NOT** supported by SQLAlchemy
656 and may change at any time.
657
658 """
659
660 supports_statement_cache = True
661 default_paramstyle = "numeric"
662 driver = "pysqlite_numeric"
663
664 _first_bind = ":1"
665 _not_in_statement_regexp = None
666
667 def __init__(self, *arg, **kw):
668 kw.setdefault("paramstyle", "numeric")
669 super().__init__(*arg, **kw)
670
671 def create_connect_args(self, url):
672 arg, opts = super().create_connect_args(url)
673 opts["factory"] = self._fix_sqlite_issue_99953()
674 return arg, opts
675
676 def _fix_sqlite_issue_99953(self):
677 import sqlite3
678
679 first_bind = self._first_bind
680 if self._not_in_statement_regexp:
681 nis = self._not_in_statement_regexp
682
683 def _test_sql(sql):
684 m = nis.search(sql)
685 assert not m, f"Found {nis.pattern!r} in {sql!r}"
686
687 else:
688
689 def _test_sql(sql):
690 pass
691
692 def _numeric_param_as_dict(parameters):
693 if parameters:
694 assert isinstance(parameters, tuple)
695 return {
696 str(idx): value for idx, value in enumerate(parameters, 1)
697 }
698 else:
699 return ()
700
701 class SQLiteFix99953Cursor(sqlite3.Cursor):
702 def execute(self, sql, parameters=()):
703 _test_sql(sql)
704 if first_bind in sql:
705 parameters = _numeric_param_as_dict(parameters)
706 return super().execute(sql, parameters)
707
708 def executemany(self, sql, parameters):
709 _test_sql(sql)
710 if first_bind in sql:
711 parameters = [
712 _numeric_param_as_dict(p) for p in parameters
713 ]
714 return super().executemany(sql, parameters)
715
716 class SQLiteFix99953Connection(sqlite3.Connection):
717 def cursor(self, factory=None):
718 if factory is None:
719 factory = SQLiteFix99953Cursor
720 return super().cursor(factory=factory)
721
722 def execute(self, sql, parameters=()):
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, parameters):
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, **kw):
755 kw.setdefault("paramstyle", "numeric_dollar")
756 super().__init__(*arg, **kw)
757