1# dialects/sqlite/pysqlcipher.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# mypy: ignore-errors
8
9
10"""
11.. dialect:: sqlite+pysqlcipher
12 :name: pysqlcipher
13 :dbapi: sqlcipher 3 or pysqlcipher
14 :connectstring: sqlite+pysqlcipher://:passphrase@/file_path[?kdf_iter=<iter>]
15
16 Dialect for support of DBAPIs that make use of the
17 `SQLCipher <https://www.zetetic.net/sqlcipher>`_ backend.
18
19
20Driver
21------
22
23Current dialect selection logic is:
24
25* If the :paramref:`_sa.create_engine.module` parameter supplies a DBAPI module,
26 that module is used.
27* Otherwise for Python 3, choose https://pypi.org/project/sqlcipher3/
28* If not available, fall back to https://pypi.org/project/pysqlcipher3/
29* For Python 2, https://pypi.org/project/pysqlcipher/ is used.
30
31.. warning:: The ``pysqlcipher3`` and ``pysqlcipher`` DBAPI drivers are no
32 longer maintained; the ``sqlcipher3`` driver as of this writing appears
33 to be current. For future compatibility, any pysqlcipher-compatible DBAPI
34 may be used as follows::
35
36 import sqlcipher_compatible_driver
37
38 from sqlalchemy import create_engine
39
40 e = create_engine(
41 "sqlite+pysqlcipher://:password@/dbname.db",
42 module=sqlcipher_compatible_driver,
43 )
44
45These drivers make use of the SQLCipher engine. This system essentially
46introduces new PRAGMA commands to SQLite which allows the setting of a
47passphrase and other encryption parameters, allowing the database file to be
48encrypted.
49
50
51Connect Strings
52---------------
53
54The format of the connect string is in every way the same as that
55of the :mod:`~sqlalchemy.dialects.sqlite.pysqlite` driver, except that the
56"password" field is now accepted, which should contain a passphrase::
57
58 e = create_engine("sqlite+pysqlcipher://:testing@/foo.db")
59
60For an absolute file path, two leading slashes should be used for the
61database name::
62
63 e = create_engine("sqlite+pysqlcipher://:testing@//path/to/foo.db")
64
65A selection of additional encryption-related pragmas supported by SQLCipher
66as documented at https://www.zetetic.net/sqlcipher/sqlcipher-api/ can be passed
67in the query string, and will result in that PRAGMA being called for each
68new connection. Currently, ``cipher``, ``kdf_iter``
69``cipher_page_size`` and ``cipher_use_hmac`` are supported::
70
71 e = create_engine(
72 "sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000"
73 )
74
75.. warning:: Previous versions of sqlalchemy did not take into consideration
76 the encryption-related pragmas passed in the url string, that were silently
77 ignored. This may cause errors when opening files saved by a
78 previous sqlalchemy version if the encryption options do not match.
79
80
81Pooling Behavior
82----------------
83
84The driver makes a change to the default pool behavior of pysqlite
85as described in :ref:`pysqlite_threading_pooling`. The pysqlcipher driver
86has been observed to be significantly slower on connection than the
87pysqlite driver, most likely due to the encryption overhead, so the
88dialect here defaults to using the :class:`.SingletonThreadPool`
89implementation,
90instead of the :class:`.NullPool` pool used by pysqlite. As always, the pool
91implementation is entirely configurable using the
92:paramref:`_sa.create_engine.poolclass` parameter; the :class:`.
93StaticPool` may
94be more feasible for single-threaded use, or :class:`.NullPool` may be used
95to prevent unencrypted connections from being held open for long periods of
96time, at the expense of slower startup time for new connections.
97
98
99""" # noqa
100
101from .pysqlite import SQLiteDialect_pysqlite
102from ... import pool
103from ... import util
104
105
106class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
107 driver = "pysqlcipher"
108 supports_statement_cache = True
109
110 pragmas = ("kdf_iter", "cipher", "cipher_page_size", "cipher_use_hmac")
111
112 @classmethod
113 def import_dbapi(cls):
114 try:
115 import sqlcipher3 as sqlcipher
116 except ImportError:
117 pass
118 else:
119 return sqlcipher
120
121 from pysqlcipher3 import dbapi2 as sqlcipher
122
123 return sqlcipher
124
125 def retrieve_dbapi_version(self, dbapi):
126 # the version of sqlcipher3 / pysqlcipher3, rather than the Python
127 # version reported by the pysqlite dialect
128 return util.parse_version_string(getattr(dbapi, "version", None))
129
130 @classmethod
131 def get_pool_class(cls, url):
132 return pool.SingletonThreadPool
133
134 def on_connect_url(self, url):
135 super_on_connect = super().on_connect_url(url)
136
137 # pull the info we need from the URL early. Even though URL
138 # is immutable, we don't want any in-place changes to the URL
139 # to affect things
140 ip = self.identifier_preparer
141 passphrase = ip.quote_identifier(url.password or "")
142 query_pragmas = {
143 prag: ip.quote_identifier(url.query[prag])
144 for prag in self.pragmas
145 if url.query.get(prag) is not None
146 }
147
148 def on_connect(conn):
149 cursor = conn.cursor()
150 cursor.execute(f"pragma key={passphrase}")
151 for prag, value in query_pragmas.items():
152 cursor.execute(f"pragma {prag}={value}")
153 cursor.close()
154
155 if super_on_connect:
156 super_on_connect(conn)
157
158 return on_connect
159
160 def create_connect_args(self, url):
161 plain_url = url._replace(password=None)
162 plain_url = plain_url.difference_update_query(self.pragmas)
163 return super().create_connect_args(plain_url)
164
165
166dialect = SQLiteDialect_pysqlcipher