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-2025 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 passphrase = url.password or "" 

135 url_query = dict(url.query) 

136 

137 def on_connect(conn): 

138 cursor = conn.cursor() 

139 cursor.execute('pragma key="%s"' % passphrase) 

140 for prag in self.pragmas: 

141 value = url_query.get(prag, None) 

142 if value is not None: 

143 cursor.execute('pragma %s="%s"' % (prag, value)) 

144 cursor.close() 

145 

146 if super_on_connect: 

147 super_on_connect(conn) 

148 

149 return on_connect 

150 

151 def create_connect_args(self, url): 

152 plain_url = url._replace(password=None) 

153 plain_url = plain_url.difference_update_query(self.pragmas) 

154 return super().create_connect_args(plain_url) 

155 

156 

157dialect = SQLiteDialect_pysqlcipher