Use UPSERT to update cache
This commit is contained in:
1 parent
c1310f677a
commit
cd1c57dddf
7 files changed
+66
-18
No files matched your search
@@ -105,6 +105,10 @@ Database
|
||||
Return if the database supports ``FOR UPDATE`` and ``FOR SHARE`` in
|
||||
``SELECT``.
|
||||
|
||||
.. method:: Database.has_insert_on_conflict()
|
||||
|
||||
Return if the database supports ``ON CONFLICT`` IN ``INSERT``.
|
||||
|
||||
.. method:: Database.get_select_for_skip_locked()
|
||||
|
||||
Return For class with skip locked.
|
||||
|
||||
+1
-1
@@ -140,7 +140,7 @@ setup(name=name,
|
||||
'Genshi',
|
||||
'python-dateutil',
|
||||
'polib',
|
||||
'python-sql >= 1.3',
|
||||
'python-sql >= 1.5',
|
||||
'werkzeug >= 0.12',
|
||||
'passlib >= 1.7.0',
|
||||
'pytz;python_version<"3.9"',
|
||||
|
||||
@@ -85,6 +85,9 @@ class DatabaseInterface(object):
|
||||
def has_select_for(self):
|
||||
return False
|
||||
|
||||
def has_insert_on_conflict(self):
|
||||
return False
|
||||
|
||||
def get_select_for_skip_locked(self):
|
||||
return For
|
||||
|
||||
|
||||
@@ -193,6 +193,7 @@ class Database(DatabaseInterface):
|
||||
_search_path = None
|
||||
_current_user = None
|
||||
_has_returning = None
|
||||
_has_insert_on_conflict = None
|
||||
_has_select_for_skip_locked = None
|
||||
_has_proc = defaultdict(lambda: defaultdict(dict))
|
||||
_extensions = defaultdict(dict)
|
||||
@@ -524,6 +525,16 @@ class Database(DatabaseInterface):
|
||||
def has_select_for(self):
|
||||
return True
|
||||
|
||||
def has_insert_on_conflict(self):
|
||||
if self._has_insert_on_conflict is None:
|
||||
connection = self.get_connection()
|
||||
try:
|
||||
self._has_insert_on_conflict = (
|
||||
self.get_version(connection) >= (9, 5))
|
||||
finally:
|
||||
self.put_connection(connection)
|
||||
return self._has_insert_on_conflict
|
||||
|
||||
def get_select_for_skip_locked(self):
|
||||
if self._has_select_for_skip_locked is None:
|
||||
connection = self.get_connection()
|
||||
|
||||
@@ -601,6 +601,9 @@ class Database(DatabaseInterface):
|
||||
def has_multirow_insert(self):
|
||||
return True
|
||||
|
||||
def has_insert_on_conflict(self):
|
||||
return sqlite.sqlite_version_info >= (3, 35, 0)
|
||||
|
||||
def has_window_functions(self):
|
||||
return sqlite.sqlite_version_info >= (3, 25, 0)
|
||||
|
||||
|
||||
+32
-17
@@ -10,7 +10,7 @@ from collections import OrderedDict, defaultdict
|
||||
from copy import deepcopy
|
||||
from weakref import WeakKeyDictionary
|
||||
|
||||
from sql import Table
|
||||
from sql import Conflict, Table
|
||||
from sql.aggregate import Max
|
||||
from sql.functions import CurrentTimestamp, Function
|
||||
|
||||
@@ -295,25 +295,40 @@ class MemoryCache(BaseCache):
|
||||
try:
|
||||
with connection.cursor() as cursor:
|
||||
for name in reset:
|
||||
cursor.execute(*table.select(table.name, table.id,
|
||||
table.timestamp,
|
||||
where=table.name == name,
|
||||
limit=1))
|
||||
if cursor.fetchone():
|
||||
# It would be better to insert only
|
||||
cursor.execute(*table.update([table.timestamp],
|
||||
[CurrentTimestamp()],
|
||||
where=table.name == name))
|
||||
if database.has_insert_on_conflict():
|
||||
query = table.insert(
|
||||
[table.timestamp, table.name],
|
||||
[[CurrentTimestamp(), name]],
|
||||
on_conflict=Conflict(
|
||||
table,
|
||||
indexed_columns=[table.name],
|
||||
columns=[table.timestamp],
|
||||
values=[CurrentTimestamp()]))
|
||||
if database.has_returning():
|
||||
query.returning = [table.timestamp]
|
||||
cursor.execute(*query)
|
||||
else:
|
||||
cursor.execute(*table.insert(
|
||||
cursor.execute(*table.select(table.name, table.id,
|
||||
table.timestamp,
|
||||
where=table.name == name,
|
||||
limit=1))
|
||||
if cursor.fetchone():
|
||||
query = table.update(
|
||||
[table.timestamp],
|
||||
[CurrentTimestamp()],
|
||||
where=table.name == name)
|
||||
else:
|
||||
query = table.insert(
|
||||
[table.timestamp, table.name],
|
||||
[[CurrentTimestamp(), name]]))
|
||||
|
||||
cursor.execute(*table.select(
|
||||
_cast(Max(table.timestamp)),
|
||||
where=table.name == name))
|
||||
[[CurrentTimestamp(), name]])
|
||||
if database.has_returning():
|
||||
query.returning = [table.timestamp]
|
||||
cursor.execute(*query)
|
||||
if not database.has_returning():
|
||||
cursor.execute(*table.select(
|
||||
Max(table.timestamp),
|
||||
where=table.name == name))
|
||||
timestamp, = cursor.fetchone()
|
||||
|
||||
inst = cls._instances[name]
|
||||
inst._clear(dbname, timestamp)
|
||||
connection.commit()
|
||||
|
||||
@@ -5,6 +5,7 @@ import datetime as dt
|
||||
import os
|
||||
import time
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from trytond import backend
|
||||
from trytond import cache as cache_mod
|
||||
@@ -224,6 +225,17 @@ class MemoryCacheTestCase(unittest.TestCase):
|
||||
self.assertEqual(cache_expire.get('foo'), None)
|
||||
|
||||
|
||||
class MemoryCacheWithoutInsertOnConflict(MemoryCacheTestCase):
|
||||
"Test Cache without Insert On Conflict"
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
patcher = patch('trytond.backend.Database.has_insert_on_conflict')
|
||||
self.addCleanup(patcher.stop)
|
||||
mock = patcher.start()
|
||||
mock.return_value = False
|
||||
|
||||
|
||||
@unittest.skipIf(backend.name == 'sqlite', "SQLite has not channel")
|
||||
class MemoryCacheChannelTestCase(MemoryCacheTestCase):
|
||||
"Test Cache with channel"
|
||||
|
||||
Reference in new issue
Block a user