We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Should the cache backend even use transactions?
mozilla-services/kinto-fxa#33 seems to be related to the fact that we're opening a transaction to consult the authentication cache. Is this really necessary? It seems like for a cache, we don't need any of the ACID properties. /cc @leplatrem 

Should the cache backend even use transactions?
mozilla-services/kinto-fxa#33 seems to be related to the fact that we're opening a transaction to consult the authentication cache. Is this really necessary? It seems like for a cache, we don't need any of the ACID properties. /cc @leplatrem 


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kinto/core/storage/postgresql/client.py`
Content:
```
1 import contextlib
2 import warnings
3 from collections import defaultdict
4 
5 from kinto.core import logger
6 from kinto.core.storage import exceptions
7 from kinto.core.utils import sqlalchemy
8 import transaction as zope_transaction
9 
10 
11 class PostgreSQLClient(object):
12     def __init__(self, session_factory, commit_manually=True, invalidate=None):
13         self.session_factory = session_factory
14         self.commit_manually = commit_manually
15         self.invalidate = invalidate or (lambda session: None)
16 
17         # # Register ujson, globally for all futur cursors
18         # with self.connect() as cursor:
19         #     psycopg2.extras.register_json(cursor,
20         #                                   globally=True,
21         #                                   loads=json.loads)
22 
23     @contextlib.contextmanager
24     def connect(self, readonly=False, force_commit=False):
25         """
26         Pulls a connection from the pool when context is entered and
27         returns it when context is exited.
28 
29         A COMMIT is performed on the current transaction if everything went
30         well. Otherwise transaction is ROLLBACK, and everything cleaned up.
31         """
32         with_transaction = not readonly and self.commit_manually
33         session = None
34         try:
35             # Pull connection from pool.
36             session = self.session_factory()
37             # Start context
38             yield session
39             if not readonly and not self.commit_manually:
40                 # Mark session as dirty.
41                 self.invalidate(session)
42             # Success
43             if with_transaction:
44                 session.commit()
45             elif force_commit:
46                 # Commit like would do a succesful request.
47                 zope_transaction.commit()
48 
49         except sqlalchemy.exc.SQLAlchemyError as e:
50             logger.error(e)
51             if session and with_transaction:
52                 session.rollback()
53             raise exceptions.BackendError(original=e)
54 
55         finally:
56             if session and self.commit_manually:
57                 # Give back to pool if commit done manually.
58                 session.close()
59 
60 
61 # Reuse existing client if same URL.
62 _CLIENTS = defaultdict(dict)
63 
64 
65 def create_from_config(config, prefix=''):
66     """Create a PostgreSQLClient client using settings in the provided config.
67     """
68     if sqlalchemy is None:
69         message = ("PostgreSQL SQLAlchemy dependency missing. "
70                    "Refer to installation section in documentation.")
71         raise ImportWarning(message)
72 
73     from zope.sqlalchemy import ZopeTransactionExtension, invalidate
74     from sqlalchemy.orm import sessionmaker, scoped_session
75 
76     settings = config.get_settings().copy()
77     # Custom Kinto settings, unsupported by SQLAlchemy.
78     settings.pop(prefix + 'backend', None)
79     settings.pop(prefix + 'max_fetch_size', None)
80     settings.pop(prefix + 'prefix', None)
81     transaction_per_request = settings.pop('transaction_per_request', False)
82 
83     url = settings[prefix + 'url']
84     existing_client = _CLIENTS[transaction_per_request].get(url)
85     if existing_client:
86         msg = ("Reuse existing PostgreSQL connection. "
87                "Parameters %s* will be ignored." % prefix)
88         warnings.warn(msg)
89         return existing_client
90 
91     # Initialize SQLAlchemy engine from settings.
92     poolclass_key = prefix + 'poolclass'
93     settings.setdefault(poolclass_key, ('kinto.core.storage.postgresql.'
94                                         'pool.QueuePoolWithMaxBacklog'))
95     settings[poolclass_key] = config.maybe_dotted(settings[poolclass_key])
96     engine = sqlalchemy.engine_from_config(settings, prefix=prefix, url=url)
97 
98     # Initialize thread-safe session factory.
99     options = {}
100     if transaction_per_request:
101         # Plug with Pyramid transaction manager
102         options['extension'] = ZopeTransactionExtension()
103     session_factory = scoped_session(sessionmaker(bind=engine, **options))
104 
105     # Store one client per URI.
106     commit_manually = (not transaction_per_request)
107     client = PostgreSQLClient(session_factory, commit_manually, invalidate)
108     _CLIENTS[transaction_per_request][url] = client
109     return client
110
```

Path: `kinto/core/cache/postgresql/__init__.py`
Content:
```
1 from __future__ import absolute_import
2 
3 import os
4 
5 from kinto.core import logger
6 from kinto.core.cache import CacheBase
7 from kinto.core.storage.postgresql.client import create_from_config
8 from kinto.core.utils import json
9 
10 
11 class Cache(CacheBase):
12     """Cache backend using PostgreSQL.
13 
14     Enable in configuration::
15 
16         kinto.cache_backend = kinto.core.cache.postgresql
17 
18     Database location URI can be customized::
19 
20         kinto.cache_url = postgres://user:pass@db.server.lan:5432/dbname
21 
22     Alternatively, username and password could also rely on system user ident
23     or even specified in :file:`~/.pgpass` (*see PostgreSQL documentation*).
24 
25     .. note::
26 
27         Some tables and indices are created when ``kinto migrate`` is run.
28         This requires some privileges on the database, or some error will
29         be raised.
30 
31         **Alternatively**, the schema can be initialized outside the
32         python application, using the SQL file located in
33         :file:`kinto/core/cache/postgresql/schema.sql`. This allows to
34         distinguish schema manipulation privileges from schema usage.
35 
36 
37     A connection pool is enabled by default::
38 
39         kinto.cache_pool_size = 10
40         kinto.cache_maxoverflow = 10
41         kinto.cache_max_backlog = -1
42         kinto.cache_pool_recycle = -1
43         kinto.cache_pool_timeout = 30
44         kinto.cache_poolclass =
45             kinto.core.storage.postgresql.pool.QueuePoolWithMaxBacklog
46 
47     The ``max_backlog``  limits the number of threads that can be in the queue
48     waiting for a connection.  Once this limit has been reached, any further
49     attempts to acquire a connection will be rejected immediately, instead of
50     locking up all threads by keeping them waiting in the queue.
51 
52     See `dedicated section in SQLAlchemy documentation
53     <http://docs.sqlalchemy.org/en/rel_1_0/core/engines.html>`_
54     for default values and behaviour.
55 
56     .. note::
57 
58         Using a `dedicated connection pool <http://pgpool.net>`_ is still
59         recommended to allow load balancing, replication or limit the number
60         of connections used in a multi-process deployment.
61 
62     :noindex:
63     """  # NOQA
64     def __init__(self, client, *args, **kwargs):
65         super(Cache, self).__init__(*args, **kwargs)
66         self.client = client
67 
68     def initialize_schema(self, dry_run=False):
69         # Check if cache table exists.
70         query = """
71         SELECT 1
72           FROM information_schema.tables
73          WHERE table_name = 'cache';
74         """
75         with self.client.connect(readonly=True) as conn:
76             result = conn.execute(query)
77             if result.rowcount > 0:
78                 logger.info("PostgreSQL cache schema is up-to-date.")
79                 return
80 
81         # Create schema
82         here = os.path.abspath(os.path.dirname(__file__))
83         sql_file = os.path.join(here, 'schema.sql')
84 
85         if dry_run:
86             logger.info("Create cache schema from %s" % sql_file)
87             return
88 
89         # Since called outside request, force commit.
90         schema = open(sql_file).read()
91         with self.client.connect(force_commit=True) as conn:
92             conn.execute(schema)
93         logger.info('Created PostgreSQL cache tables')
94 
95     def flush(self):
96         query = """
97         DELETE FROM cache;
98         """
99         # Since called outside request (e.g. tests), force commit.
100         with self.client.connect(force_commit=True) as conn:
101             conn.execute(query)
102         logger.debug('Flushed PostgreSQL cache tables')
103 
104     def ttl(self, key):
105         query = """
106         SELECT EXTRACT(SECOND FROM (ttl - now())) AS ttl
107           FROM cache
108          WHERE key = :key
109            AND ttl IS NOT NULL;
110         """
111         with self.client.connect(readonly=True) as conn:
112             result = conn.execute(query, dict(key=self.prefix + key))
113             if result.rowcount > 0:
114                 return result.fetchone()['ttl']
115         return -1
116 
117     def expire(self, key, ttl):
118         query = """
119         UPDATE cache SET ttl = sec2ttl(:ttl) WHERE key = :key;
120         """
121         with self.client.connect() as conn:
122             conn.execute(query, dict(ttl=ttl, key=self.prefix + key))
123 
124     def set(self, key, value, ttl=None):
125         if ttl is None:
126             logger.warning("No TTL for cache key %r" % key)
127         query = """
128         WITH upsert AS (
129             UPDATE cache SET value = :value, ttl = sec2ttl(:ttl)
130              WHERE key=:key
131             RETURNING *)
132         INSERT INTO cache (key, value, ttl)
133         SELECT :key, :value, sec2ttl(:ttl)
134         WHERE NOT EXISTS (SELECT * FROM upsert)
135         """
136         value = json.dumps(value)
137         with self.client.connect() as conn:
138             conn.execute(query, dict(key=self.prefix + key,
139                                      value=value, ttl=ttl))
140 
141     def get(self, key):
142         purge = "DELETE FROM cache WHERE ttl IS NOT NULL AND now() > ttl;"
143         query = "SELECT value FROM cache WHERE key = :key;"
144         with self.client.connect() as conn:
145             conn.execute(purge)
146             result = conn.execute(query, dict(key=self.prefix + key))
147             if result.rowcount > 0:
148                 value = result.fetchone()['value']
149                 return json.loads(value)
150 
151     def delete(self, key):
152         query = "DELETE FROM cache WHERE key = :key"
153         with self.client.connect() as conn:
154             conn.execute(query, dict(key=self.prefix + key))
155 
156 
157 def load_from_config(config):
158     settings = config.get_settings()
159     client = create_from_config(config, prefix='cache_')
160     return Cache(client=client, cache_prefix=settings['cache_prefix'])
161
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/kinto/core/cache/postgresql/__init__.py b/kinto/core/cache/postgresql/__init__.py
--- a/kinto/core/cache/postgresql/__init__.py
+++ b/kinto/core/cache/postgresql/__init__.py
@@ -156,5 +156,5 @@
 
 def load_from_config(config):
     settings = config.get_settings()
-    client = create_from_config(config, prefix='cache_')
+    client = create_from_config(config, prefix='cache_', with_transaction=False)
     return Cache(client=client, cache_prefix=settings['cache_prefix'])
diff --git a/kinto/core/storage/postgresql/client.py b/kinto/core/storage/postgresql/client.py
--- a/kinto/core/storage/postgresql/client.py
+++ b/kinto/core/storage/postgresql/client.py
@@ -62,7 +62,7 @@
 _CLIENTS = defaultdict(dict)
 
 
-def create_from_config(config, prefix=''):
+def create_from_config(config, prefix='', with_transaction=True):
     """Create a PostgreSQLClient client using settings in the provided config.
     """
     if sqlalchemy is None:
@@ -77,8 +77,9 @@
     # Custom Kinto settings, unsupported by SQLAlchemy.
     settings.pop(prefix + 'backend', None)
     settings.pop(prefix + 'max_fetch_size', None)
+    settings.pop(prefix + 'max_size_bytes', None)
     settings.pop(prefix + 'prefix', None)
-    transaction_per_request = settings.pop('transaction_per_request', False)
+    transaction_per_request = with_transaction and settings.pop('transaction_per_request', False)
 
     url = settings[prefix + 'url']
     existing_client = _CLIENTS[transaction_per_request].get(url)

{"golden_diff": "diff --git a/kinto/core/cache/postgresql/__init__.py b/kinto/core/cache/postgresql/__init__.py\n--- a/kinto/core/cache/postgresql/__init__.py\n+++ b/kinto/core/cache/postgresql/__init__.py\n@@ -156,5 +156,5 @@\n \n def load_from_config(config):\n     settings = config.get_settings()\n-    client = create_from_config(config, prefix='cache_')\n+    client = create_from_config(config, prefix='cache_', with_transaction=False)\n     return Cache(client=client, cache_prefix=settings['cache_prefix'])\ndiff --git a/kinto/core/storage/postgresql/client.py b/kinto/core/storage/postgresql/client.py\n--- a/kinto/core/storage/postgresql/client.py\n+++ b/kinto/core/storage/postgresql/client.py\n@@ -62,7 +62,7 @@\n _CLIENTS = defaultdict(dict)\n \n \n-def create_from_config(config, prefix=''):\n+def create_from_config(config, prefix='', with_transaction=True):\n     \"\"\"Create a PostgreSQLClient client using settings in the provided config.\n     \"\"\"\n     if sqlalchemy is None:\n@@ -77,8 +77,9 @@\n     # Custom Kinto settings, unsupported by SQLAlchemy.\n     settings.pop(prefix + 'backend', None)\n     settings.pop(prefix + 'max_fetch_size', None)\n+    settings.pop(prefix + 'max_size_bytes', None)\n     settings.pop(prefix + 'prefix', None)\n-    transaction_per_request = settings.pop('transaction_per_request', False)\n+    transaction_per_request = with_transaction and settings.pop('transaction_per_request', False)\n \n     url = settings[prefix + 'url']\n     existing_client = _CLIENTS[transaction_per_request].get(url)\n", "issue": "Should the cache backend even use transactions?\nmozilla-services/kinto-fxa#33 seems to be related to the fact that we're opening a transaction to consult the authentication cache. Is this really necessary? It seems like for a cache, we don't need any of the ACID properties. /cc @leplatrem \n\nShould the cache backend even use transactions?\nmozilla-services/kinto-fxa#33 seems to be related to the fact that we're opening a transaction to consult the authentication cache. Is this really necessary? It seems like for a cache, we don't need any of the ACID properties. /cc @leplatrem \n\n", "before_files": [{"content": "import contextlib\nimport warnings\nfrom collections import defaultdict\n\nfrom kinto.core import logger\nfrom kinto.core.storage import exceptions\nfrom kinto.core.utils import sqlalchemy\nimport transaction as zope_transaction\n\n\nclass PostgreSQLClient(object):\n    def __init__(self, session_factory, commit_manually=True, invalidate=None):\n        self.session_factory = session_factory\n        self.commit_manually = commit_manually\n        self.invalidate = invalidate or (lambda session: None)\n\n        # # Register ujson, globally for all futur cursors\n        # with self.connect() as cursor:\n        #     psycopg2.extras.register_json(cursor,\n        #                                   globally=True,\n        #                                   loads=json.loads)\n\n    @contextlib.contextmanager\n    def connect(self, readonly=False, force_commit=False):\n        \"\"\"\n        Pulls a connection from the pool when context is entered and\n        returns it when context is exited.\n\n        A COMMIT is performed on the current transaction if everything went\n        well. Otherwise transaction is ROLLBACK, and everything cleaned up.\n        \"\"\"\n        with_transaction = not readonly and self.commit_manually\n        session = None\n        try:\n            # Pull connection from pool.\n            session = self.session_factory()\n            # Start context\n            yield session\n            if not readonly and not self.commit_manually:\n                # Mark session as dirty.\n                self.invalidate(session)\n            # Success\n            if with_transaction:\n                session.commit()\n            elif force_commit:\n                # Commit like would do a succesful request.\n                zope_transaction.commit()\n\n        except sqlalchemy.exc.SQLAlchemyError as e:\n            logger.error(e)\n            if session and with_transaction:\n                session.rollback()\n            raise exceptions.BackendError(original=e)\n\n        finally:\n            if session and self.commit_manually:\n                # Give back to pool if commit done manually.\n                session.close()\n\n\n# Reuse existing client if same URL.\n_CLIENTS = defaultdict(dict)\n\n\ndef create_from_config(config, prefix=''):\n    \"\"\"Create a PostgreSQLClient client using settings in the provided config.\n    \"\"\"\n    if sqlalchemy is None:\n        message = (\"PostgreSQL SQLAlchemy dependency missing. \"\n                   \"Refer to installation section in documentation.\")\n        raise ImportWarning(message)\n\n    from zope.sqlalchemy import ZopeTransactionExtension, invalidate\n    from sqlalchemy.orm import sessionmaker, scoped_session\n\n    settings = config.get_settings().copy()\n    # Custom Kinto settings, unsupported by SQLAlchemy.\n    settings.pop(prefix + 'backend', None)\n    settings.pop(prefix + 'max_fetch_size', None)\n    settings.pop(prefix + 'prefix', None)\n    transaction_per_request = settings.pop('transaction_per_request', False)\n\n    url = settings[prefix + 'url']\n    existing_client = _CLIENTS[transaction_per_request].get(url)\n    if existing_client:\n        msg = (\"Reuse existing PostgreSQL connection. \"\n               \"Parameters %s* will be ignored.\" % prefix)\n        warnings.warn(msg)\n        return existing_client\n\n    # Initialize SQLAlchemy engine from settings.\n    poolclass_key = prefix + 'poolclass'\n    settings.setdefault(poolclass_key, ('kinto.core.storage.postgresql.'\n                                        'pool.QueuePoolWithMaxBacklog'))\n    settings[poolclass_key] = config.maybe_dotted(settings[poolclass_key])\n    engine = sqlalchemy.engine_from_config(settings, prefix=prefix, url=url)\n\n    # Initialize thread-safe session factory.\n    options = {}\n    if transaction_per_request:\n        # Plug with Pyramid transaction manager\n        options['extension'] = ZopeTransactionExtension()\n    session_factory = scoped_session(sessionmaker(bind=engine, **options))\n\n    # Store one client per URI.\n    commit_manually = (not transaction_per_request)\n    client = PostgreSQLClient(session_factory, commit_manually, invalidate)\n    _CLIENTS[transaction_per_request][url] = client\n    return client\n", "path": "kinto/core/storage/postgresql/client.py"}, {"content": "from __future__ import absolute_import\n\nimport os\n\nfrom kinto.core import logger\nfrom kinto.core.cache import CacheBase\nfrom kinto.core.storage.postgresql.client import create_from_config\nfrom kinto.core.utils import json\n\n\nclass Cache(CacheBase):\n    \"\"\"Cache backend using PostgreSQL.\n\n    Enable in configuration::\n\n        kinto.cache_backend = kinto.core.cache.postgresql\n\n    Database location URI can be customized::\n\n        kinto.cache_url = postgres://user:pass@db.server.lan:5432/dbname\n\n    Alternatively, username and password could also rely on system user ident\n    or even specified in :file:`~/.pgpass` (*see PostgreSQL documentation*).\n\n    .. note::\n\n        Some tables and indices are created when ``kinto migrate`` is run.\n        This requires some privileges on the database, or some error will\n        be raised.\n\n        **Alternatively**, the schema can be initialized outside the\n        python application, using the SQL file located in\n        :file:`kinto/core/cache/postgresql/schema.sql`. This allows to\n        distinguish schema manipulation privileges from schema usage.\n\n\n    A connection pool is enabled by default::\n\n        kinto.cache_pool_size = 10\n        kinto.cache_maxoverflow = 10\n        kinto.cache_max_backlog = -1\n        kinto.cache_pool_recycle = -1\n        kinto.cache_pool_timeout = 30\n        kinto.cache_poolclass =\n            kinto.core.storage.postgresql.pool.QueuePoolWithMaxBacklog\n\n    The ``max_backlog``  limits the number of threads that can be in the queue\n    waiting for a connection.  Once this limit has been reached, any further\n    attempts to acquire a connection will be rejected immediately, instead of\n    locking up all threads by keeping them waiting in the queue.\n\n    See `dedicated section in SQLAlchemy documentation\n    <http://docs.sqlalchemy.org/en/rel_1_0/core/engines.html>`_\n    for default values and behaviour.\n\n    .. note::\n\n        Using a `dedicated connection pool <http://pgpool.net>`_ is still\n        recommended to allow load balancing, replication or limit the number\n        of connections used in a multi-process deployment.\n\n    :noindex:\n    \"\"\"  # NOQA\n    def __init__(self, client, *args, **kwargs):\n        super(Cache, self).__init__(*args, **kwargs)\n        self.client = client\n\n    def initialize_schema(self, dry_run=False):\n        # Check if cache table exists.\n        query = \"\"\"\n        SELECT 1\n          FROM information_schema.tables\n         WHERE table_name = 'cache';\n        \"\"\"\n        with self.client.connect(readonly=True) as conn:\n            result = conn.execute(query)\n            if result.rowcount > 0:\n                logger.info(\"PostgreSQL cache schema is up-to-date.\")\n                return\n\n        # Create schema\n        here = os.path.abspath(os.path.dirname(__file__))\n        sql_file = os.path.join(here, 'schema.sql')\n\n        if dry_run:\n            logger.info(\"Create cache schema from %s\" % sql_file)\n            return\n\n        # Since called outside request, force commit.\n        schema = open(sql_file).read()\n        with self.client.connect(force_commit=True) as conn:\n            conn.execute(schema)\n        logger.info('Created PostgreSQL cache tables')\n\n    def flush(self):\n        query = \"\"\"\n        DELETE FROM cache;\n        \"\"\"\n        # Since called outside request (e.g. tests), force commit.\n        with self.client.connect(force_commit=True) as conn:\n            conn.execute(query)\n        logger.debug('Flushed PostgreSQL cache tables')\n\n    def ttl(self, key):\n        query = \"\"\"\n        SELECT EXTRACT(SECOND FROM (ttl - now())) AS ttl\n          FROM cache\n         WHERE key = :key\n           AND ttl IS NOT NULL;\n        \"\"\"\n        with self.client.connect(readonly=True) as conn:\n            result = conn.execute(query, dict(key=self.prefix + key))\n            if result.rowcount > 0:\n                return result.fetchone()['ttl']\n        return -1\n\n    def expire(self, key, ttl):\n        query = \"\"\"\n        UPDATE cache SET ttl = sec2ttl(:ttl) WHERE key = :key;\n        \"\"\"\n        with self.client.connect() as conn:\n            conn.execute(query, dict(ttl=ttl, key=self.prefix + key))\n\n    def set(self, key, value, ttl=None):\n        if ttl is None:\n            logger.warning(\"No TTL for cache key %r\" % key)\n        query = \"\"\"\n        WITH upsert AS (\n            UPDATE cache SET value = :value, ttl = sec2ttl(:ttl)\n             WHERE key=:key\n            RETURNING *)\n        INSERT INTO cache (key, value, ttl)\n        SELECT :key, :value, sec2ttl(:ttl)\n        WHERE NOT EXISTS (SELECT * FROM upsert)\n        \"\"\"\n        value = json.dumps(value)\n        with self.client.connect() as conn:\n            conn.execute(query, dict(key=self.prefix + key,\n                                     value=value, ttl=ttl))\n\n    def get(self, key):\n        purge = \"DELETE FROM cache WHERE ttl IS NOT NULL AND now() > ttl;\"\n        query = \"SELECT value FROM cache WHERE key = :key;\"\n        with self.client.connect() as conn:\n            conn.execute(purge)\n            result = conn.execute(query, dict(key=self.prefix + key))\n            if result.rowcount > 0:\n                value = result.fetchone()['value']\n                return json.loads(value)\n\n    def delete(self, key):\n        query = \"DELETE FROM cache WHERE key = :key\"\n        with self.client.connect() as conn:\n            conn.execute(query, dict(key=self.prefix + key))\n\n\ndef load_from_config(config):\n    settings = config.get_settings()\n    client = create_from_config(config, prefix='cache_')\n    return Cache(client=client, cache_prefix=settings['cache_prefix'])\n", "path": "kinto/core/cache/postgresql/__init__.py"}], "after_files": [{"content": "import contextlib\nimport warnings\nfrom collections import defaultdict\n\nfrom kinto.core import logger\nfrom kinto.core.storage import exceptions\nfrom kinto.core.utils import sqlalchemy\nimport transaction as zope_transaction\n\n\nclass PostgreSQLClient(object):\n    def __init__(self, session_factory, commit_manually=True, invalidate=None):\n        self.session_factory = session_factory\n        self.commit_manually = commit_manually\n        self.invalidate = invalidate or (lambda session: None)\n\n        # # Register ujson, globally for all futur cursors\n        # with self.connect() as cursor:\n        #     psycopg2.extras.register_json(cursor,\n        #                                   globally=True,\n        #                                   loads=json.loads)\n\n    @contextlib.contextmanager\n    def connect(self, readonly=False, force_commit=False):\n        \"\"\"\n        Pulls a connection from the pool when context is entered and\n        returns it when context is exited.\n\n        A COMMIT is performed on the current transaction if everything went\n        well. Otherwise transaction is ROLLBACK, and everything cleaned up.\n        \"\"\"\n        with_transaction = not readonly and self.commit_manually\n        session = None\n        try:\n            # Pull connection from pool.\n            session = self.session_factory()\n            # Start context\n            yield session\n            if not readonly and not self.commit_manually:\n                # Mark session as dirty.\n                self.invalidate(session)\n            # Success\n            if with_transaction:\n                session.commit()\n            elif force_commit:\n                # Commit like would do a succesful request.\n                zope_transaction.commit()\n\n        except sqlalchemy.exc.SQLAlchemyError as e:\n            logger.error(e)\n            if session and with_transaction:\n                session.rollback()\n            raise exceptions.BackendError(original=e)\n\n        finally:\n            if session and self.commit_manually:\n                # Give back to pool if commit done manually.\n                session.close()\n\n\n# Reuse existing client if same URL.\n_CLIENTS = defaultdict(dict)\n\n\ndef create_from_config(config, prefix='', with_transaction=True):\n    \"\"\"Create a PostgreSQLClient client using settings in the provided config.\n    \"\"\"\n    if sqlalchemy is None:\n        message = (\"PostgreSQL SQLAlchemy dependency missing. \"\n                   \"Refer to installation section in documentation.\")\n        raise ImportWarning(message)\n\n    from zope.sqlalchemy import ZopeTransactionExtension, invalidate\n    from sqlalchemy.orm import sessionmaker, scoped_session\n\n    settings = config.get_settings().copy()\n    # Custom Kinto settings, unsupported by SQLAlchemy.\n    settings.pop(prefix + 'backend', None)\n    settings.pop(prefix + 'max_fetch_size', None)\n    settings.pop(prefix + 'max_size_bytes', None)\n    settings.pop(prefix + 'prefix', None)\n    transaction_per_request = with_transaction and settings.pop('transaction_per_request', False)\n\n    url = settings[prefix + 'url']\n    existing_client = _CLIENTS[transaction_per_request].get(url)\n    if existing_client:\n        msg = (\"Reuse existing PostgreSQL connection. \"\n               \"Parameters %s* will be ignored.\" % prefix)\n        warnings.warn(msg)\n        return existing_client\n\n    # Initialize SQLAlchemy engine from settings.\n    poolclass_key = prefix + 'poolclass'\n    settings.setdefault(poolclass_key, ('kinto.core.storage.postgresql.'\n                                        'pool.QueuePoolWithMaxBacklog'))\n    settings[poolclass_key] = config.maybe_dotted(settings[poolclass_key])\n    engine = sqlalchemy.engine_from_config(settings, prefix=prefix, url=url)\n\n    # Initialize thread-safe session factory.\n    options = {}\n    if transaction_per_request:\n        # Plug with Pyramid transaction manager\n        options['extension'] = ZopeTransactionExtension()\n    session_factory = scoped_session(sessionmaker(bind=engine, **options))\n\n    # Store one client per URI.\n    commit_manually = (not transaction_per_request)\n    client = PostgreSQLClient(session_factory, commit_manually, invalidate)\n    _CLIENTS[transaction_per_request][url] = client\n    return client\n", "path": "kinto/core/storage/postgresql/client.py"}, {"content": "from __future__ import absolute_import\n\nimport os\n\nfrom kinto.core import logger\nfrom kinto.core.cache import CacheBase\nfrom kinto.core.storage.postgresql.client import create_from_config\nfrom kinto.core.utils import json\n\n\nclass Cache(CacheBase):\n    \"\"\"Cache backend using PostgreSQL.\n\n    Enable in configuration::\n\n        kinto.cache_backend = kinto.core.cache.postgresql\n\n    Database location URI can be customized::\n\n        kinto.cache_url = postgres://user:pass@db.server.lan:5432/dbname\n\n    Alternatively, username and password could also rely on system user ident\n    or even specified in :file:`~/.pgpass` (*see PostgreSQL documentation*).\n\n    .. note::\n\n        Some tables and indices are created when ``kinto migrate`` is run.\n        This requires some privileges on the database, or some error will\n        be raised.\n\n        **Alternatively**, the schema can be initialized outside the\n        python application, using the SQL file located in\n        :file:`kinto/core/cache/postgresql/schema.sql`. This allows to\n        distinguish schema manipulation privileges from schema usage.\n\n\n    A connection pool is enabled by default::\n\n        kinto.cache_pool_size = 10\n        kinto.cache_maxoverflow = 10\n        kinto.cache_max_backlog = -1\n        kinto.cache_pool_recycle = -1\n        kinto.cache_pool_timeout = 30\n        kinto.cache_poolclass =\n            kinto.core.storage.postgresql.pool.QueuePoolWithMaxBacklog\n\n    The ``max_backlog``  limits the number of threads that can be in the queue\n    waiting for a connection.  Once this limit has been reached, any further\n    attempts to acquire a connection will be rejected immediately, instead of\n    locking up all threads by keeping them waiting in the queue.\n\n    See `dedicated section in SQLAlchemy documentation\n    <http://docs.sqlalchemy.org/en/rel_1_0/core/engines.html>`_\n    for default values and behaviour.\n\n    .. note::\n\n        Using a `dedicated connection pool <http://pgpool.net>`_ is still\n        recommended to allow load balancing, replication or limit the number\n        of connections used in a multi-process deployment.\n\n    :noindex:\n    \"\"\"  # NOQA\n    def __init__(self, client, *args, **kwargs):\n        super(Cache, self).__init__(*args, **kwargs)\n        self.client = client\n\n    def initialize_schema(self, dry_run=False):\n        # Check if cache table exists.\n        query = \"\"\"\n        SELECT 1\n          FROM information_schema.tables\n         WHERE table_name = 'cache';\n        \"\"\"\n        with self.client.connect(readonly=True) as conn:\n            result = conn.execute(query)\n            if result.rowcount > 0:\n                logger.info(\"PostgreSQL cache schema is up-to-date.\")\n                return\n\n        # Create schema\n        here = os.path.abspath(os.path.dirname(__file__))\n        sql_file = os.path.join(here, 'schema.sql')\n\n        if dry_run:\n            logger.info(\"Create cache schema from %s\" % sql_file)\n            return\n\n        # Since called outside request, force commit.\n        schema = open(sql_file).read()\n        with self.client.connect(force_commit=True) as conn:\n            conn.execute(schema)\n        logger.info('Created PostgreSQL cache tables')\n\n    def flush(self):\n        query = \"\"\"\n        DELETE FROM cache;\n        \"\"\"\n        # Since called outside request (e.g. tests), force commit.\n        with self.client.connect(force_commit=True) as conn:\n            conn.execute(query)\n        logger.debug('Flushed PostgreSQL cache tables')\n\n    def ttl(self, key):\n        query = \"\"\"\n        SELECT EXTRACT(SECOND FROM (ttl - now())) AS ttl\n          FROM cache\n         WHERE key = :key\n           AND ttl IS NOT NULL;\n        \"\"\"\n        with self.client.connect(readonly=True) as conn:\n            result = conn.execute(query, dict(key=self.prefix + key))\n            if result.rowcount > 0:\n                return result.fetchone()['ttl']\n        return -1\n\n    def expire(self, key, ttl):\n        query = \"\"\"\n        UPDATE cache SET ttl = sec2ttl(:ttl) WHERE key = :key;\n        \"\"\"\n        with self.client.connect() as conn:\n            conn.execute(query, dict(ttl=ttl, key=self.prefix + key))\n\n    def set(self, key, value, ttl=None):\n        if ttl is None:\n            logger.warning(\"No TTL for cache key %r\" % key)\n        query = \"\"\"\n        WITH upsert AS (\n            UPDATE cache SET value = :value, ttl = sec2ttl(:ttl)\n             WHERE key=:key\n            RETURNING *)\n        INSERT INTO cache (key, value, ttl)\n        SELECT :key, :value, sec2ttl(:ttl)\n        WHERE NOT EXISTS (SELECT * FROM upsert)\n        \"\"\"\n        value = json.dumps(value)\n        with self.client.connect() as conn:\n            conn.execute(query, dict(key=self.prefix + key,\n                                     value=value, ttl=ttl))\n\n    def get(self, key):\n        purge = \"DELETE FROM cache WHERE ttl IS NOT NULL AND now() > ttl;\"\n        query = \"SELECT value FROM cache WHERE key = :key;\"\n        with self.client.connect() as conn:\n            conn.execute(purge)\n            result = conn.execute(query, dict(key=self.prefix + key))\n            if result.rowcount > 0:\n                value = result.fetchone()['value']\n                return json.loads(value)\n\n    def delete(self, key):\n        query = \"DELETE FROM cache WHERE key = :key\"\n        with self.client.connect() as conn:\n            conn.execute(query, dict(key=self.prefix + key))\n\n\ndef load_from_config(config):\n    settings = config.get_settings()\n    client = create_from_config(config, prefix='cache_', with_transaction=False)\n    return Cache(client=client, cache_prefix=settings['cache_prefix'])\n", "path": "kinto/core/cache/postgresql/__init__.py"}]}