| 464 | |
| 465 | |
| 466 | class SQLiteDialect_pysqlite(SQLiteDialect): |
| 467 | default_paramstyle = "qmark" |
| 468 | supports_statement_cache = True |
| 469 | returns_native_bytes = True |
| 470 | |
| 471 | colspecs = util.update_copy( |
| 472 | SQLiteDialect.colspecs, |
| 473 | { |
| 474 | sqltypes.Date: _SQLite_pysqliteDate, |
| 475 | sqltypes.TIMESTAMP: _SQLite_pysqliteTimeStamp, |
| 476 | }, |
| 477 | ) |
| 478 | |
| 479 | description_encoding = None |
| 480 | |
| 481 | driver = "pysqlite" |
| 482 | |
| 483 | @classmethod |
| 484 | def import_dbapi(cls) -> DBAPIModule: |
| 485 | from sqlite3 import dbapi2 as sqlite |
| 486 | |
| 487 | return cast("DBAPIModule", sqlite) |
| 488 | |
| 489 | @classmethod |
| 490 | def _is_url_file_db(cls, url: URL) -> bool: |
| 491 | if (url.database and url.database != ":memory:") and ( |
| 492 | url.query.get("mode", None) != "memory" |
| 493 | ): |
| 494 | return True |
| 495 | else: |
| 496 | return False |
| 497 | |
| 498 | @classmethod |
| 499 | def get_pool_class(cls, url: URL) -> type[pool.Pool]: |
| 500 | if cls._is_url_file_db(url): |
| 501 | return pool.QueuePool |
| 502 | else: |
| 503 | return pool.SingletonThreadPool |
| 504 | |
| 505 | def _get_server_version_info(self, connection: Any) -> VersionInfoType: |
| 506 | return self.dbapi.sqlite_version_info # type: ignore |
| 507 | |
| 508 | _isolation_lookup = SQLiteDialect._isolation_lookup.union( |
| 509 | { |
| 510 | "AUTOCOMMIT": None, |
| 511 | } |
| 512 | ) |
| 513 | |
| 514 | def set_isolation_level( |
| 515 | self, dbapi_connection: DBAPIConnection, level: IsolationLevel |
| 516 | ) -> None: |
| 517 | if level == "AUTOCOMMIT": |
| 518 | dbapi_connection.isolation_level = None |
| 519 | else: |
| 520 | dbapi_connection.isolation_level = "" |
| 521 | return super().set_isolation_level(dbapi_connection, level) |
| 522 | |
| 523 | def detect_autocommit_setting(self, dbapi_conn: DBAPIConnection) -> bool: |