1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
require 'sqlite3'
module Moneta
module Adapters
# Sqlite3 backend
# @api public
class Sqlite < Adapter
include IncrementSupport
supports :create, :each_key
config :table, default: 'moneta'
config :busy_timeout, default: 1000
config :journal_mode
backend { |file:| ::SQLite3::Database.new(file) }
# @param [Hash] options
# @option options [String] :file Database file
# @option options [String] :table ('moneta') Table name
# @option options [Integer] :busy_timeout (1000) Sqlite timeout if database is busy
# @option options [::Sqlite3::Database] :backend Use existing backend instance
# @option options [String, Symbol] :journal_mode Set the journal mode for the connection
def initialize(options = {})
super
backend.busy_timeout(config.busy_timeout)
backend.execute("create table if not exists #{config.table} (k blob not null primary key, v blob)")
if journal_mode = config.journal_mode
backend.journal_mode = journal_mode.to_s
end
@stmts =
[@exists = backend.prepare("select exists(select 1 from #{config.table} where k = ?)"),
@select = backend.prepare("select v from #{config.table} where k = ?"),
@replace = backend.prepare("replace into #{config.table} values (?, ?)"),
@delete = backend.prepare("delete from #{config.table} where k = ?"),
@clear = backend.prepare("delete from #{config.table}"),
@create = backend.prepare("insert into #{config.table} values (?, ?)"),
@keys = backend.prepare("select k from #{config.table}"),
@count = backend.prepare("select count(*) from #{config.table}")]
version = backend.execute("select sqlite_version()").first.first
if @can_upsert = ::Gem::Version.new(version) >= ::Gem::Version.new('3.24.0')
@stmts << (@increment = backend.prepare <<-SQL)
insert into #{config.table} values (?, ?)
on conflict (k)
do update set v = cast(cast(v as integer) + ? as blob)
where v = '0' or v = X'30' or cast(v as integer) != 0
SQL
end
end
# (see Proxy#key?)
def key?(key, options = {})
@exists.execute!(key).first.first.to_i == 1
end
# (see Proxy#load)
def load(key, options = {})
rows = @select.execute!(key)
rows.empty? ? nil : rows.first.first
end
# (see Proxy#store)
def store(key, value, options = {})
@replace.execute!(key, value)
value
end
# (see Proxy#delete)
def delete(key, options = {})
value = load(key, options)
@delete.execute!(key)
value
end
# (see Proxy#increment)
def increment(key, amount = 1, options = {})
backend.transaction(:exclusive) { return super } unless @can_upsert
backend.transaction do
@increment.execute!(key, amount.to_s, amount)
return Integer(load(key))
end
end
# (see Proxy#clear)
def clear(options = {})
@clear.execute!
self
end
# (see Default#create)
def create(key, value, options = {})
@create.execute!(key, value)
true
rescue SQLite3::ConstraintException
# If you know a better way to detect whether an insert-ignore
# suceeded, please tell me.
@create.reset!
false
end
# (see Proxy#close)
def close
@stmts.each { |s| s.close }
backend.close
nil
end
# (see Proxy#slice)
def slice(*keys, **options)
query = "select k, v from #{config.table} where k in (#{(['?'] * keys.length).join(',')})"
backend.execute(query, keys)
end
# (see Proxy#values_at)
def values_at(*keys, **options)
hash = Hash[slice(*keys, **options)]
keys.map { |key| hash[key] }
end
# (see Proxy#fetch_values)
def fetch_values(*keys, **options)
return values_at(*keys, **options) unless block_given?
hash = Hash[slice(*keys, **options)]
keys.map do |key|
if hash.key?(key)
hash[key]
else
yield key
end
end
end
# (see Proxy#merge!)
def merge!(pairs, options = {})
transaction = backend.transaction if block_given?
if block_given?
existing = Hash[slice(*pairs.map { |k, _| k }.to_a)]
pairs = pairs.map do |key, new_value|
new_value = yield(key, existing[key], new_value) if existing.key?(key)
[key, new_value]
end.to_a
else
pairs = pairs.to_a
end
unless pairs.empty?
query = "replace into #{config.table} (k, v) values" + (['(?, ?)'] * pairs.length).join(',')
backend.query(query, pairs.flatten).close
end
rescue
backend.rollback if transaction
raise
else
backend.commit if transaction
self
end
# (see Proxy#each_key)
def each_key
return enum_for(:each_key) { @count.execute!.first.first } unless block_given?
@keys.execute!.each do |row|
yield row.first
end
self
end
end
end
end
|
