Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/dialects/sqlite/pysqlcipher.py : 38%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sqlite/pysqlcipher.py
2# Copyright (C) 2005-2020 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
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
15 ``pysqlcipher`` is a fork of the standard ``pysqlite`` driver to make
16 use of the `SQLCipher <https://www.zetetic.net/sqlcipher>`_ backend.
18 ``pysqlcipher3`` is a fork of ``pysqlcipher`` for Python 3. This dialect
19 will attempt to import it if ``pysqlcipher`` is non-present.
21 .. versionadded:: 1.1.4 - added fallback import for pysqlcipher3
23 .. versionadded:: 0.9.9 - added pysqlcipher dialect
25Driver
26------
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.
35`pysqlcipher3` is a fork of `pysqlcipher` with support for Python 3,
36the driver is the same.
38Connect Strings
39---------------
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::
45 e = create_engine('sqlite+pysqlcipher://:testing@/foo.db')
47For an absolute file path, two leading slashes should be used for the
48database name::
50 e = create_engine('sqlite+pysqlcipher://:testing@//path/to/foo.db')
52A selection of additional encryption-related pragmas supported by SQLCipher
53as documented at https://www.zetetic.net/sqlcipher/sqlcipher-api/ can be passed
54in the query string, and will result in that PRAGMA being called for each
55new connection. Currently, ``cipher``, ``kdf_iter``
56``cipher_page_size`` and ``cipher_use_hmac`` are supported::
58 e = create_engine('sqlite+pysqlcipher://:testing@/foo.db?cipher=aes-256-cfb&kdf_iter=64000')
61Pooling Behavior
62----------------
64The driver makes a change to the default pool behavior of pysqlite
65as described in :ref:`pysqlite_threading_pooling`. The pysqlcipher driver
66has been observed to be significantly slower on connection than the
67pysqlite driver, most likely due to the encryption overhead, so the
68dialect here defaults to using the :class:`.SingletonThreadPool`
69implementation,
70instead of the :class:`.NullPool` pool used by pysqlite. As always, the pool
71implementation is entirely configurable using the
72:paramref:`_sa.create_engine.poolclass` parameter; the :class:`.StaticPool`
73may
74be more feasible for single-threaded use, or :class:`.NullPool` may be used
75to prevent unencrypted connections from being held open for long periods of
76time, at the expense of slower startup time for new connections.
79""" # noqa
81from __future__ import absolute_import
83from .pysqlite import SQLiteDialect_pysqlite
84from ... import pool
85from ...engine import url as _url
88class SQLiteDialect_pysqlcipher(SQLiteDialect_pysqlite):
89 driver = "pysqlcipher"
91 pragmas = ("kdf_iter", "cipher", "cipher_page_size", "cipher_use_hmac")
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
104 @classmethod
105 def get_pool_class(cls, url):
106 return pool.SingletonThreadPool
108 def connect(self, *cargs, **cparams):
109 passphrase = cparams.pop("passphrase", "")
111 pragmas = dict((key, cparams.pop(key, None)) for key in self.pragmas)
113 conn = super(SQLiteDialect_pysqlcipher, self).connect(
114 *cargs, **cparams
115 )
116 conn.execute('pragma key="%s"' % passphrase)
117 for prag, value in pragmas.items():
118 if value is not None:
119 conn.execute('pragma %s="%s"' % (prag, value))
121 return conn
123 def create_connect_args(self, url):
124 super_url = _url.URL(
125 url.drivername,
126 username=url.username,
127 host=url.host,
128 database=url.database,
129 query=url.query,
130 )
131 c_args, opts = super(
132 SQLiteDialect_pysqlcipher, self
133 ).create_connect_args(super_url)
134 opts["passphrase"] = url.password
135 return c_args, opts
138dialect = SQLiteDialect_pysqlcipher