Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/sqlite/pysqlcipher.py: 38%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

42 statements  

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