Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def write(self, query_builder, **kwargs):
query = query_builder(**kwargs).query
config.logger.debug(query)
self.db.execute(*query)
row_ids = self.db.lexicon.row_ids(self.db, kwargs['primary_key'])
if len(row_ids) > 0:
return self.select(where={kwargs['primary_key']: row_ids})
return pandas.DataFrame(columns=self.columns())
def insert(self, **kwargs):
"""
Performs an INSERT statement on the model's table in the master database.
:param values: A dictionary containing the values to be inserted. ``datetime``, ``dict`` and ``bool`` objects can be passed as is and will be correctly serialized by psycopg2.
:type values: dict
"""
if len(kwargs['values']) == 0:
config.logger.warning('No values to insert.')
return
values = kwargs['values']
if isinstance(values, self):
values = values.attributes.copy()
if isinstance(values, dict):
for (k, v) in list(values.items()):
if v is None:
del kwargs['values'][k]
kwargs['stack'] = self.stack_mark(inspect.stack())
kwargs['primary_key'] = self.primary_key
column_names = self.table_schema().keys()
now = datetime.utcnow()
for field in [f for f in ('created_at', 'updated_at') if f in column_names]:
if isinstance(kwargs['values'], dict):
def connect(self):
connection = super(DatabaseConnection, self).connect()
connection.initialize(config.logger)
return connection