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

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 

103 

104 

105class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite): 

106 driver = "pysqlcipher" 

107 supports_statement_cache = True 

108 

109 pragmas = ("kdf_iter", "cipher", "cipher_page_size", "cipher_use_hmac") 

110 

111 @classmethod 

112 def import_dbapi(cls): 

113 try: 

114 import sqlcipher3 as sqlcipher 

115 except ImportError: 

116 pass 

117 else: 

118 return sqlcipher 

119 

120 from pysqlcipher3 import dbapi2 as sqlcipher 

121 

122 return sqlcipher 

123 

124 @classmethod 

125 def get_pool_class(cls, url): 

126 return pool.SingletonThreadPool 

127 

128 def on_connect_url(self, url): 

129 super_on_connect = super().on_connect_url(url) 

130 

131 # pull the info we need from the URL early. Even though URL 

132 # is immutable, we don't want any in-place changes to the URL 

133 # to affect things 

134 ip = self.identifier_preparer 

135 passphrase = ip.quote_identifier(url.password or "") 

136 query_pragmas = { 

137 prag: ip.quote_identifier(url.query[prag]) 

138 for prag in self.pragmas 

139 if url.query.get(prag) is not None 

140 } 

141 

142 def on_connect(conn): 

143 cursor = conn.cursor() 

144 cursor.execute(f"pragma key={passphrase}") 

145 for prag, value in query_pragmas.items(): 

146 cursor.execute(f"pragma {prag}={value}") 

147 print(query_pragmas) 

148 cursor.close() 

149 

150 if super_on_connect: 

151 super_on_connect(conn) 

152 

153 return on_connect 

154 

155 def create_connect_args(self, url): 

156 plain_url = url._replace(password=None) 

157 plain_url = plain_url.difference_update_query(self.pragmas) 

158 return super().create_connect_args(plain_url) 

159 

160 

161dialect = SQLiteDialect_pysqlcipher