| 583 | |
| 584 | |
| 585 | class SQLiteEngine(Engine[SQLiteTransaction]): |
| 586 | __slots__ = ("connection_kwargs",) |
| 587 | |
| 588 | def __init__( |
| 589 | self, |
| 590 | path: str = "piccolo.sqlite", |
| 591 | log_queries: bool = False, |
| 592 | log_responses: bool = False, |
| 593 | **connection_kwargs, |
| 594 | ) -> None: |
| 595 | """ |
| 596 | :param path: |
| 597 | A relative or absolute path to the the SQLite database file (it |
| 598 | will be created if it doesn't already exist). |
| 599 | :param log_queries: |
| 600 | If ``True``, all SQL and DDL statements are printed out before |
| 601 | being run. Useful for debugging. |
| 602 | :param log_responses: |
| 603 | If ``True``, the raw response from each query is printed out. |
| 604 | Useful for debugging. |
| 605 | :param connection_kwargs: |
| 606 | These are passed directly to the database adapter. We recommend |
| 607 | setting ``timeout`` if you expect your application to process a |
| 608 | large number of concurrent writes, to prevent queries timing out. |
| 609 | See Python's `sqlite3 docs <https://docs.python.org/3/library/sqlite3.html#sqlite3.connect>`_ |
| 610 | for more info. |
| 611 | |
| 612 | """ # noqa: E501 |
| 613 | default_connection_kwargs = { |
| 614 | "database": path, |
| 615 | "detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES, |
| 616 | "isolation_level": None, |
| 617 | } |
| 618 | |
| 619 | self.log_queries = log_queries |
| 620 | self.log_responses = log_responses |
| 621 | self.connection_kwargs = { |
| 622 | **default_connection_kwargs, |
| 623 | **connection_kwargs, |
| 624 | } |
| 625 | |
| 626 | self.current_transaction = contextvars.ContextVar( |
| 627 | f"sqlite_current_transaction_{path}", default=None |
| 628 | ) |
| 629 | |
| 630 | super().__init__( |
| 631 | engine_type="sqlite", |
| 632 | min_version_number=3.25, |
| 633 | log_queries=log_queries, |
| 634 | log_responses=log_responses, |
| 635 | ) |
| 636 | |
| 637 | @property |
| 638 | def path(self): |
| 639 | return self.connection_kwargs["database"] |
| 640 | |
| 641 | @path.setter |
| 642 | def path(self, value: str): |
no outgoing calls