remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-24 19:04:12 +02:00
Code Issues Projects Releases Packages Wiki Activity

Extract build_insert_queries_and_params

Browse source
This commit is contained in:
Simon Wiles 2020-09-07 16:01:25 -07:00
commit c241162fa9
1 changed files with 106 additions and 79 deletions
Download patch file Download diff file Expand all files Collapse all files

185
sqlite_utils/db.py
View file

@ -1031,6 +1031,100 @@ class Table(Queryable):
self.last_pk = pk_values[0] if len(self.pks) == 1 else pk_values self.last_pk = pk_values[0] if len(self.pks) == 1 else pk_values
return self return self
def build_insert_queries_and_params(
self,
extracts,
chunk,
all_columns,
hash_id,
upsert,
pk,
conversions,
num_records_processed,
replace,
ignore,
):
# values is the list of insert data that is passed to the
# .execute() method - but some of them may be replaced by
# new primary keys if we are extracting any columns.
values = []
extracts = resolve_extracts(extracts)
for record in chunk:
record_values = []
for key in all_columns:
value = jsonify_if_needed(
record.get(key, None if key != hash_id else _hash(record))
)
if key in extracts:
extract_table = extracts[key]
value = self.db[extract_table].lookup({"value": value})
record_values.append(value)
values.append(record_values)
queries_and_params = []
if upsert:
if isinstance(pk, str):
pks = [pk]
else:
pks = pk
self.last_pk = None
for record_values in values:
# TODO: make more efficient:
record = dict(zip(all_columns, record_values))
sql = "INSERT OR IGNORE INTO [{table}]({pks}) VALUES({pk_placeholders});".format(
table=self.name,
pks=", ".join(["[{}]".format(p) for p in pks]),
pk_placeholders=", ".join(["?" for p in pks]),
)
queries_and_params.append((sql, [record[col] for col in pks]))
# UPDATE [book] SET [name] = 'Programming' WHERE [id] = 1001;
set_cols = [col for col in all_columns if col not in pks]
sql2 = "UPDATE [{table}] SET {pairs} WHERE {wheres}".format(
table=self.name,
pairs=", ".join(
"[{}] = {}".format(col, conversions.get(col, "?"))
for col in set_cols
),
wheres=" AND ".join("[{}] = ?".format(pk) for pk in pks),
)
queries_and_params.append(
(
sql2,
[record[col] for col in set_cols] + [record[pk] for pk in pks],
)
)
# We can populate .last_pk right here
if num_records_processed == 1:
self.last_pk = tuple(record[pk] for pk in pks)
if len(self.last_pk) == 1:
self.last_pk = self.last_pk[0]
else:
or_what = ""
if replace:
or_what = "OR REPLACE "
elif ignore:
or_what = "OR IGNORE "
sql = """
INSERT {or_what}INTO [{table}] ({columns}) VALUES {rows};
""".strip().format(
or_what=or_what,
table=self.name,
columns=", ".join("[{}]".format(c) for c in all_columns),
rows=", ".join(
"({placeholders})".format(
placeholders=", ".join(
[conversions.get(col, "?") for col in all_columns]
)
)
for record in chunk
),
)
flat_values = list(itertools.chain(*values))
queries_and_params = [(sql, flat_values)]
return queries_and_params
def insert( def insert(
self, self,
record, record,
@ -1164,86 +1258,19 @@ class Table(Queryable):
validate_column_names(all_columns) validate_column_names(all_columns)
first = False first = False
# values is the list of insert data that is passed to the
# .execute() method - but some of them may be replaced by
# new primary keys if we are extracting any columns.
values = []
extracts = resolve_extracts(extracts)
for record in chunk:
record_values = []
for key in all_columns:
value = jsonify_if_needed(
record.get(key, None if key != hash_id else _hash(record))
)
if key in extracts:
extract_table = extracts[key]
value = self.db[extract_table].lookup({"value": value})
record_values.append(value)
values.append(record_values)
queries_and_params = [] queries_and_params = self.build_insert_queries_and_params(
if upsert: extracts,
if isinstance(pk, str): chunk,
pks = [pk] all_columns,
else: hash_id,
pks = pk upsert,
self.last_pk = None pk,
for record_values in values: conversions,
# TODO: make more efficient: num_records_processed,
record = dict(zip(all_columns, record_values)) replace,
params = [] ignore,
sql = "INSERT OR IGNORE INTO [{table}]({pks}) VALUES({pk_placeholders});".format( )
table=self.name,
pks=", ".join(["[{}]".format(p) for p in pks]),
pk_placeholders=", ".join(["?" for p in pks]),
)
queries_and_params.append((sql, [record[col] for col in pks]))
# UPDATE [book] SET [name] = 'Programming' WHERE [id] = 1001;
set_cols = [col for col in all_columns if col not in pks]
sql2 = "UPDATE [{table}] SET {pairs} WHERE {wheres}".format(
table=self.name,
pairs=", ".join(
"[{}] = {}".format(col, conversions.get(col, "?"))
for col in set_cols
),
wheres=" AND ".join("[{}] = ?".format(pk) for pk in pks),
)
queries_and_params.append(
(
sql2,
[record[col] for col in set_cols]
+ [record[pk] for pk in pks],
)
)
# We can populate .last_pk right here
if num_records_processed == 1:
self.last_pk = tuple(record[pk] for pk in pks)
if len(self.last_pk) == 1:
self.last_pk = self.last_pk[0]
else:
or_what = ""
if replace:
or_what = "OR REPLACE "
elif ignore:
or_what = "OR IGNORE "
sql = """
INSERT {or_what}INTO [{table}] ({columns}) VALUES {rows};
""".strip().format(
or_what=or_what,
table=self.name,
columns=", ".join("[{}]".format(c) for c in all_columns),
rows=", ".join(
"({placeholders})".format(
placeholders=", ".join(
[conversions.get(col, "?") for col in all_columns]
)
)
for record in chunk
),
)
flat_values = list(itertools.chain(*values))
queries_and_params = [(sql, flat_values)]
with self.db.conn: with self.db.conn:
for query, params in queries_and_params: for query, params in queries_and_params: