Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/SQLAlchemy-1.3.25.dev0-py3.11-linux-x86_64.egg/sqlalchemy/dialects/sqlite/pysqlcipher.py: 44%

1# sqlite/pysqlcipher.py 

2# Copyright (C) 2005-2021 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: http://www.opensource.org/licenses/mit-license.php 

7 

8""" 

9.. dialect:: sqlite+pysqlcipher 

10 :name: pysqlcipher 

11 :dbapi: pysqlcipher 

12 :connectstring: sqlite+pysqlcipher://:passphrase/file_path[?kdf_iter=<iter>] 

13 :url: https://pypi.python.org/pypi/pysqlcipher 

14 

15 ``pysqlcipher`` is a fork of the standard ``pysqlite`` driver to make 

16 use of the `SQLCipher <https://www.zetetic.net/sqlcipher>`_ backend. 

17 

18 ``pysqlcipher3`` is a fork of ``pysqlcipher`` for Python 3. This dialect 

19 will attempt to import it if ``pysqlcipher`` is non-present. 

20 

21 .. versionadded:: 1.1.4 - added fallback import for pysqlcipher3 

22 

23 .. versionadded:: 0.9.9 - added pysqlcipher dialect 

24 

25Driver 

26------ 

27 

28The driver here is the 

29`pysqlcipher <https://pypi.python.org/pypi/pysqlcipher>`_ 

30driver, which makes use of the SQLCipher engine. This system essentially 

31introduces new PRAGMA commands to SQLite which allows the setting of a 

32passphrase and other encryption parameters, allowing the database 

33file to be encrypted. 

34 

35`pysqlcipher3` is a fork of `pysqlcipher` with support for Python 3, 

36the driver is the same. 

37 

38Connect Strings 

39--------------- 

40 

41The format of the connect string is in every way the same as that 

42of the :mod:`~sqlalchemy.dialects.sqlite.pysqlite` driver, except that the 

43"password" field is now accepted, which should contain a passphrase:: 

44 

45 e = create_engine('sqlite+pysqlcipher://:testing@/foo.db') 

46 

47For an absolute file path, two leading slashes should be used for the 

48database name:: 

49 

50 e = create_engine('sqlite+pysqlcipher://:testing@//path/to/foo.db') 

51 

52Additional encryption-related pragmas must be executed manually, 

53using the ``first_connect`` pool event. A selection of the pragmas supported 

54by SQLCipher is documented at 

55https://www.zetetic.net/sqlcipher/sqlcipher-api/. 

56 

57.. warning:: Previously the documentation wrongly stated that these 

58 pragma could be passed in the url string. This has never worked 

59 for the 1.3 series of sqlalchemy. The 1.4 series adds proper 

60 support for them when passed in the url string. 

61 

62 

63Pooling Behavior 

64---------------- 

65 

66The driver makes a change to the default pool behavior of pysqlite 

67as described in :ref:`pysqlite_threading_pooling`. The pysqlcipher driver 

68has been observed to be significantly slower on connection than the 

69pysqlite driver, most likely due to the encryption overhead, so the 

70dialect here defaults to using the :class:`.SingletonThreadPool` 

71implementation, 

72instead of the :class:`.NullPool` pool used by pysqlite. As always, the pool 

73implementation is entirely configurable using the 

74:paramref:`_sa.create_engine.poolclass` parameter; the :class:`.StaticPool` 

75may 

76be more feasible for single-threaded use, or :class:`.NullPool` may be used 

77to prevent unencrypted connections from being held open for long periods of 

78time, at the expense of slower startup time for new connections. 

79 

80 

81""" # noqa 

82 

83from __future__ import absolute_import 

84 

85from .pysqlite import SQLiteDialect_pysqlite 

86from ... import pool 

87from ...engine import url as _url 

88 

89 

90class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite): 

91 driver = "pysqlcipher" 

92 

93 @classmethod 

94 def dbapi(cls): 

95 try: 

96 from pysqlcipher import dbapi2 as sqlcipher 

97 except ImportError as e: 

98 try: 

99 from pysqlcipher3 import dbapi2 as sqlcipher 

100 except ImportError: 

101 raise e 

102 return sqlcipher 

103 

104 @classmethod 

105 def get_pool_class(cls, url): 

106 return pool.SingletonThreadPool 

107 

108 def connect(self, *cargs, **cparams): 

109 passphrase = cparams.pop("passphrase", "") 

110 

111 conn = super(SQLiteDialect_pysqlcipher, self).connect( 

112 *cargs, **cparams 

113 ) 

114 conn.execute('pragma key="%s"' % passphrase) 

115 

116 return conn 

117 

118 def create_connect_args(self, url): 

119 super_url = _url.URL( 

120 url.drivername, 

121 username=url.username, 

122 host=url.host, 

123 database=url.database, 

124 query=url.query, 

125 ) 

126 c_args, opts = super( 

127 SQLiteDialect_pysqlcipher, self 

128 ).create_connect_args(super_url) 

129 opts["passphrase"] = url.password 

130 return c_args, opts 

131 

132 

133dialect = SQLiteDialect_pysqlcipher