Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'INSERT INTO',
self.table_name,
self.list_wrap(self.fields),
'VALUES',
', '.join(
[self.list_wrap(ext) for ext in self.value_extrapolators]
)
]
if self.scheme == 'postgres':
query += ['RETURNING', self.primary_key]
query = ' '.join(query) + ';'
query = self.apply_watermark(query)
return (query, self.values)
class UpdateQueryBuilder(WriteQueryBuilder, SelectQueryBuilder):
@memoized_property
def query(self):
query = 'UPDATE ' + self.table_name + ' SET '
query += self.lexicon.update_values(self.fields, self.value_extrapolators)
if self.wheres: query += " WHERE " + self.wheres
if self.scheme == 'postgres':
query += ' RETURNING ' + self.primary_key
query += ';'
query = self.apply_watermark(query)
values = self.where_values
values.update(self.values)
return (query, values)
def query(self):
query = 'UPDATE ' + self.table_name + ' SET '
query += self.lexicon.update_values(self.fields, self.value_extrapolators)
if self.wheres: query += " WHERE " + self.wheres
if self.scheme == 'postgres':
query += ' RETURNING ' + self.primary_key
query += ';'
query = self.apply_watermark(query)
values = self.where_values
values.update(self.values)
return (query, values)
class DeleteQueryBuilder(WriteQueryBuilder, SelectQueryBuilder):
@memoized_property
def query(self):
query = ' '.join(['DELETE', 'FROM', self.table_name, 'WHERE', self.wheres]) + ';'
query = self.apply_watermark(query)
return (query, self.lexicon.format_args(self.where_values))
class RawQueryBuilder(WriteQueryBuilder, SelectQueryBuilder):
def where_key(self, key):
return key
@memoized_property
def sql(self):
if 'sql' in self.kwargs and self.kwargs['sql']:
query = self.apply_watermark(query)
values = self.where_values
values.update(self.values)
return (query, values)
class DeleteQueryBuilder(WriteQueryBuilder, SelectQueryBuilder):
@memoized_property
def query(self):
query = ' '.join(['DELETE', 'FROM', self.table_name, 'WHERE', self.wheres]) + ';'
query = self.apply_watermark(query)
return (query, self.lexicon.format_args(self.where_values))
class RawQueryBuilder(WriteQueryBuilder, SelectQueryBuilder):
def where_key(self, key):
return key
@memoized_property
def sql(self):
if 'sql' in self.kwargs and self.kwargs['sql']:
raw_sql = self.kwargs['sql']
if 'filename' in self.kwargs and self.kwargs['filename']:
with open(self.kwargs['filename']) as file:
raw_sql = file.read()
return raw_sql
@memoized_property
def query(self):
query = self.apply_watermark(self.sql)
def watermark(self):
if self.scheme == 'sqlite':
return ''
else:
return super(WriteQueryBuilder, self).watermark
values.update(**v)
return values
@memoized_property
def fields(self):
return self.write_values.columns
@memoized_property
def watermark(self):
if self.scheme == 'sqlite':
return ''
else:
return super(WriteQueryBuilder, self).watermark
class InsertQueryBuilder(WriteQueryBuilder):
@memoized_property
def query(self):
query = [
'INSERT INTO',
self.table_name,
self.list_wrap(self.fields),
'VALUES',
', '.join(
[self.list_wrap(ext) for ext in self.value_extrapolators]
)
]
if self.scheme == 'postgres':
query += ['RETURNING', self.primary_key]
query = ' '.join(query) + ';'
query = self.apply_watermark(query)