KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use update_columns method in pandera
Best Python code snippet using pandera_python
db_connection.py
Source:db_connection.py 
1import os2import json3from collections import defaultdict, OrderedDict4class BaseDBConnection(object):5 """ Base KG connection for database6 """7 def __init__(self, db_path, chunksize):8 """ Create an connection to database9 :param db_path: database path10 :type db_path: str11 :param chunksize: the chunksize to load/write database12 :type chunksize: int13 """14 self._conn = None15 self.chunksize = chunksize16 def close(self):17 """ Close the connection safely18 """19 raise NotImplementedError20 def __del__(self):21 self.close()22 def create_table(self, table_name, columns, column_types):23 """ Create a table with given columns and types24 :param table_name: the table name to create25 :type table_name: str26 :param columns: the columns to create27 :type columns: List[str]28 :param column_types: the corresponding column types29 :type column_types: List[str]30 """31 raise NotImplementedError32 def get_columns(self, table_name, columns):33 """ Get column information from a table34 :param table_name: the table name to retrieve35 :type table_name: str36 :param columns: the columns to retrieve37 :type columns: List[str]38 :return: a list of retrieved rows39 :rtype: List[Dict[str, object]]40 """41 raise NotImplementedError42 def select_row(self, table_name, _id, columns):43 """ Select a row from a table44 :param table_name: the table name to retrieve45 :type table_name: str46 :param _id: the row id47 :type _id: str48 :param columns: the columns to retrieve49 :type columns: List[str]50 :return: a retrieved row51 :rtype: Dict[str, object]52 """53 raise NotImplementedError54 def select_rows(self, table_name, _ids, columns):55 """ Select rows from a table56 :param table_name: the table name to retrieve57 :type table_name: str58 :param _ids: the row ids59 :type _ids: List[str]60 :param columns: the columns to retrieve61 :type columns: List[str]62 :return: retrieved rows63 :rtype: List[Dict[str, object]]64 """65 raise NotImplementedError66 def insert_row(self, table_name, row):67 """ Insert a row into a table68 :param table_name: the table name to insert69 :type table_name: str70 :param row: the row to insert71 :type row: Dict[str, object]72 """73 raise NotImplementedError74 def insert_rows(self, table_name, rows):75 """ Insert several rows into a table76 :param table_name: the table name to insert77 :type table_name: str78 :param rows: the rows to insert79 :type rows: List[Dict[str, object]]80 """81 raise NotImplementedError82 def get_update_op(self, update_columns, operator):83 """ Get an update operator based on columns and a operator84 :param update_columns: a list of columns to update85 :type update_columns: List[str]86 :param operator: an operator that applies to the columns, including "+", "-", "*", "/", "="87 :type operator: str88 :return: an operator that suits the backend database89 :rtype: object90 """91 raise NotImplementedError92 def update_row(self, table_name, row, update_op, update_columns):93 """ Update a row that exists in a table94 :param table_name: the table name to update95 :type table_name: str96 :param row: a new row97 :type row: Dict[str, object]98 :param update_op: an operator that returned by `get_update_op`99 :type update_op: object100 :param update_columns: the columns to update101 :type update_columns: List[str]102 """103 raise NotImplementedError104 def update_rows(self, table_name, rows, update_ops, update_columns):105 """ Update rows that exist in a table106 :param table_name: the table name to update107 :type table_name: str108 :param rows: new rows109 :type rows: List[Dict[str, object]]110 :param update_ops: operator(s) that returned by `get_update_op`111 :type update_ops: Union[List[object], object]112 :param update_columns: the columns to update113 :type update_columns: List[str]114 """115 raise NotImplementedError116 def get_rows_by_keys(self, table_name, bys, keys, columns, order_bys=None, reverse=False, top_n=None):117 """ Retrieve rows by specific keys in some order118 :param table_name: the table name to retrieve119 :type table_name: str120 :param bys: the given columns to match121 :type bys: List[str]122 :param keys: the given values to match123 :type keys: List[str]124 :param columns: the given columns to retrieve125 :type columns: List[str]126 :param order_bys: the columns whose value are used to sort rows127 :type order_bys: List[str]128 :param reverse: whether to sort in a reversed order129 :type reverse: bool130 :param top_n: how many rows to return, default `None` for all rows131 :type top_n: int132 :return: retrieved rows133 :rtype: List[Dict[str, object]]134 """135 raise NotImplementedError136class SqliteDBConnection(BaseDBConnection):137 """ KG connection for SQLite database138 """139 def __init__(self, db_path, chunksize):140 """ Create an connection to SQLite database141 :param db_path: database path, e.g., /home/xliucr/ASER/KG.db142 :type db_path: str143 :param chunksize: the chunksize to load/write database144 :type chunksize: int145 """146 import sqlite3147 super(SqliteDBConnection, self).__init__(db_path, chunksize)148 self._conn = sqlite3.connect(db_path)149 def close(self):150 """ Close the connection safely151 """152 if self._conn:153 self._conn.close()154 def create_table(self, table_name, columns, column_types):155 """ Create a table with given columns and types156 :param table_name: the table name to create157 :type table_name: str158 :param columns: the columns to create159 :type columns: List[str]160 :param column_types: the corresponding column types, please refer to https://www.sqlite.org/datatype3.html161 :type column_types: List[str]162 """163 create_table = "CREATE TABLE %s (%s);" % (164 table_name, ",".join([' '.join(x) for x in zip(columns, column_types)])165 )166 self._conn.execute(create_table)167 self._conn.commit()168 def get_columns(self, table_name, columns):169 """ Get column information from a table170 :param table_name: the table name to retrieve171 :type table_name: str172 :param columns: the columns to retrieve173 :type columns: List[str]174 :return: a list of retrieved rows175 :rtype: List[Dict[str, object]]176 """177 select_table = "SELECT %s FROM %s;" % (",".join(columns), table_name)178 result = list(map(lambda x: OrderedDict(zip(columns, x)), self._conn.execute(select_table)))179 return result180 def select_row(self, table_name, _id, columns):181 """ Select a row from a table182 (suggestion: consider to use `select_rows` if you want to retrieve multiple rows)183 :param table_name: the table name to retrieve184 :type table_name: str185 :param _id: the row id186 :type _id: str187 :param columns: the columns to retrieve188 :type columns: List[str]189 :return: a retrieved row190 :rtype: Dict[str, object]191 """192 select_table = "SELECT %s FROM %s WHERE _id=?;" % (",".join(columns), table_name)193 result = list(self._conn.execute(select_table, [_id]))194 if len(result) == 0:195 return None196 else:197 return OrderedDict(zip(columns, result[0]))198 def select_rows(self, table_name, _ids, columns):199 """ Select rows from a table200 :param table_name: the table name to retrieve201 :type table_name: str202 :param _ids: the row ids203 :type _ids: List[str]204 :param columns: the columns to retrieve205 :type columns: List[str]206 :return: retrieved rows207 :rtype: List[Dict[str, object]]208 """209 if len(_ids) > 0:210 row_cache = dict()211 result = []212 for idx in range(0, len(_ids), self.chunksize):213 select_table = "SELECT %s FROM %s WHERE _id IN ('%s');" % (214 ",".join(columns), table_name, "','".join(_ids[idx:idx + self.chunksize])215 )216 result.extend(list(self._conn.execute(select_table)))217 for x in result:218 exact_match_row = OrderedDict(zip(columns, x))219 row_cache[exact_match_row["_id"]] = exact_match_row220 exact_match_rows = []221 for _id in _ids:222 exact_match_rows.append(row_cache.get(_id, None))223 return exact_match_rows224 else:225 return []226 def insert_row(self, table_name, row):227 """ Insert a row into a table228 (suggestion: consider to use `insert_rows` if you want to insert multiple rows)229 :param table_name: the table name to insert230 :type table_name: str231 :param row: the row to insert232 :type row: Dict[str, object]233 """234 insert_table = "INSERT INTO %s VALUES (%s)" % (table_name, ",".join(['?'] * (len(row))))235 self._conn.execute(insert_table, list(row.values()))236 self._conn.commit()237 def insert_rows(self, table_name, rows):238 """ Insert several rows into a table239 :param table_name: the table name to insert240 :type table_name: str241 :param rows: the rows to insert242 :type rows: List[Dict[str, object]]243 """244 if len(rows) > 0:245 insert_table = "INSERT INTO %s VALUES (%s)" % (table_name, ",".join(['?'] * (len(next(iter(rows))))))246 self._conn.executemany(insert_table, [list(row.values()) for row in rows])247 self._conn.commit()248 def get_update_op(self, update_columns, operator):249 """ Get an update operator based on columns and a operator250 :param update_columns: a list of columns to update251 :type update_columns: List[str]252 :param operator: an operator that applies to the columns, including "+", "-", "*", "/", "="253 :type operator: str254 :return: an operator that suits the backend database255 :rtype: str256 """257 if operator in "+-*/":258 update_ops = []259 for update_column in update_columns:260 update_ops.append(update_column + "=" + update_column + operator + "?")261 return ",".join(update_ops)262 elif operator == "=":263 update_ops = []264 for update_column in update_columns:265 update_ops.append(update_column + "=?")266 return ",".join(update_ops)267 else:268 raise NotImplementedError269 def _update_update_op(self, row, update_op, update_columns):270 update_op_sp = update_op.split('?')271 while len(update_op_sp) >= 0 and update_op_sp[-1] == '':272 update_op_sp.pop()273 assert len(update_op_sp) == len(update_columns)274 new_update_op = []275 for i in range(len(update_op_sp)):276 new_update_op.append(update_op_sp[i])277 if isinstance(row[update_columns[i]], str):278 new_update_op.append("'" + row[update_columns[i]].replace("'", "''") + "'")279 else:280 new_update_op.append(str(row[update_columns[i]]))281 return ''.join(new_update_op)282 def update_row(self, table_name, row, update_op, update_columns):283 """ Update a row that exists in a table284 (suggestion: consider to use `update_rows` if you want to update multiple rows)285 :param table_name: the table name to update286 :type table_name: str287 :param row: a new row288 :type row: Dict[str, object]289 :param update_op: an operator that returned by `get_update_op`290 :type update_op: str291 :param update_columns: the columns to update292 :type update_columns: List[str]293 """294 update_table = "UPDATE %s SET %s WHERE _id=?" % (table_name, update_op)295 self._conn.execute(update_table, [row[k] for k in update_columns] + [row["_id"]])296 self._conn.commit()297 def update_rows(self, table_name, rows, update_ops, update_columns):298 """ Update rows that exist in a table299 :param table_name: the table name to update300 :type table_name: str301 :param rows: new rows302 :type rows: List[Dict[str, object]]303 :param update_ops: operator(s) that returned by `get_update_op`304 :type update_ops: Union[List[str], str]305 :param update_columns: the columns to update306 :type update_columns: List[str]307 """308 if len(rows) > 0:309 if isinstance(update_ops, (tuple, list)): # +-*/310 assert len(rows) == len(update_ops)311 # group rows by op to speed up312 update_op_collections = defaultdict(list) # key: _update_update_op313 for i, row in enumerate(rows):314 # self.update_row(row, table_name, update_ops[i], update_columns)315 new_update_op = self._update_update_op(row, update_ops[i], update_columns)316 update_op_collections[new_update_op].append(row)317 for new_update_op, op_rows in update_op_collections.items():318 _ids = [row["_id"] for row in op_rows]319 for idx in range(0, len(_ids), self.chunksize):320 update_table = "UPDATE %s SET %s WHERE _id IN ('%s');" % (321 table_name, new_update_op, "','".join(_ids[idx:idx + self.chunksize])322 )323 self._conn.execute(update_table)324 else: # =325 update_op = update_ops326 # group rows by new values to speed up327 value_collections = defaultdict(list) # key: values of new values328 for row in rows:329 # self.update_row(row, table_name, update_op, update_columns)330 value_collections[json.dumps([row[k] for k in update_columns])].append(row)331 for new_update_op, op_rows in value_collections.items():332 new_update_op = self._update_update_op(op_rows[0], update_op, update_columns)333 _ids = [row["_id"] for row in op_rows]334 for idx in range(0, len(_ids), self.chunksize):335 update_table = "UPDATE %s SET %s WHERE _id IN ('%s');" % (336 table_name, new_update_op, "','".join(_ids[idx:idx + self.chunksize])337 )338 self._conn.execute(update_table)339 self._conn.commit()340 def get_rows_by_keys(self, table_name, bys, keys, columns, order_bys=None, reverse=False, top_n=None):341 """ Retrieve rows by specific keys in some order342 :param table_name: the table name to retrieve343 :type table_name: str344 :param bys: the given columns to match345 :type bys: List[str]346 :param keys: the given values to match347 :type keys: List[str]348 :param columns: the given columns to retrieve349 :type columns: List[str]350 :param order_bys: the columns whose value are used to sort rows351 :type order_bys: List[str]352 :param reverse: whether to sort in a reversed order353 :type reverse: bool354 :param top_n: how many rows to return, default `None` for all rows355 :type top_n: int356 :return: retrieved rows357 :rtype: List[Dict[str, object]]358 """359 key_match_events = []360 select_table = "SELECT %s FROM %s WHERE %s" % (361 ",".join(columns), table_name, " AND ".join(["%s=?" % (by) for by in bys])362 )363 if order_bys:364 select_table += " ORDER BY %s %s" % (",".join(order_bys), "DESC" if reverse else "ASC")365 if top_n:366 select_table += " LIMIT %d" % (top_n)367 select_table += ";"368 for x in self._conn.execute(select_table, keys):369 key_match_event = OrderedDict(zip(columns, x))370 key_match_events.append(key_match_event)371 return key_match_events372class MongoDBConnection(BaseDBConnection):373 """ KG connection for MongoDB374 """375 def __init__(self, db_path, chunksize):376 """ Create an connection to SQLite database377 :param db_path: database path, e.g., mongodb://localhost:27017/ASER378 :type db_path: str379 :param chunksize: the chunksize to load/write database380 :type chunksize: int381 """382 import pymongo383 super(MongoDBConnection, self).__init__(db_path, chunksize)384 host_port, db_name = os.path.split(db_path)385 self._client = pymongo.MongoClient(host_port, document_class=OrderedDict)386 self._conn = self._client[db_name]387 def close(self):388 """ Close the connection safely389 """390 self._client.close()391 def create_table(self, table_name):392 """ Create a table without the necessary to provide column information393 :param table_name: the table name to create394 :type table_name: str395 """396 self._conn[table_name]397 def __get_projection(self, columns):398 projection = {"_id": 0}399 for k in columns:400 projection[k] = 1401 return projection402 def get_columns(self, table_name, columns):403 """ Get column information from a table404 :param table_name: the table name to retrieve405 :type table_name: str406 :param columns: the columns to retrieve407 :type columns: List[str]408 :return: a list of retrieved rows409 :rtype: List[Dict[str, object]]410 """411 projection = self.__get_projection(columns)412 results = list(self._conn[table_name].find({}, projection))413 return results414 def select_row(self, table_name, _id, columns):415 """ Select a row from a table416 (suggestion: consider to use `select_rows` if you want to retrieve multiple rows)417 :param table_name: the table name to retrieve418 :type table_name: str419 :param _id: the row id420 :type _id: str421 :param columns: the columns to retrieve422 :type columns: List[str]423 :return: a retrieved row424 :rtype: Dict[str, object]425 """426 projection = self.__get_projection(columns)427 return self._conn[table_name].find_one({"_id": _id}, projection)428 def select_rows(self, table_name, _ids, columns):429 """ Select rows from a table430 :param table_name: the table name to retrieve431 :type table_name: str432 :param _ids: the row ids433 :type _ids: List[str]434 :param columns: the columns to retrieve435 :type columns: List[str]436 :return: retrieved rows437 :rtype: List[Dict[str, object]]438 """439 table = self._conn[table_name]440 exact_match_rows = []441 projection = self.__get_projection(columns)442 for idx in range(0, len(_ids), self.chunksize):443 query = {"_id": {'$in': _ids[idx:idx + self.chunksize]}}444 exact_match_rows.extend(table.find(query, projection))445 row_cache = {x["_id"]: x for x in exact_match_rows}446 exact_match_rows = [row_cache.get(_id, None) for _id in _ids]447 return exact_match_rows448 def insert_row(self, table_name, row):449 """ Insert a row into a table450 (suggestion: consider to use `insert_rows` if you want to insert multiple rows)451 :param table_name: the table name to insert452 :type table_name: str453 :param row: the row to insert454 :type row: Dict[str, object]455 """456 self._conn[table_name].insert_one(row)457 def insert_rows(self, table_name, rows):458 """ Insert several rows into a table459 :param table_name: the table name to insert460 :type table_name: str461 :param rows: the rows to insert462 :type rows: List[Dict[str, object]]463 """464 self._conn[table_name].insert_many(rows)465 def get_update_op(self, update_columns, operator):466 """ Get an update operator based on columns and a operator467 :param update_columns: a list of columns to update468 :type update_columns: List[str]469 :param operator: an operator that applies to the columns, including "+", "-", "*", "/", "="470 :type operator: str471 :return: an operator that suits the backend database472 :rtype: Dict[str, Dict[str, float]]473 """474 if operator == "+":475 update_ops = {}476 for update_column in update_columns:477 update_ops[update_column] = 1 # placeholder478 return {"$inc": update_ops}479 elif operator == "-":480 update_ops = {}481 for update_column in update_columns:482 update_ops[update_column] = -1 # placeholder483 return {"$inc": update_ops}484 elif operator == "*":485 update_ops = {}486 for update_column in update_columns:487 update_ops[update_column] = 2 # placeholder488 return {"$mul": update_ops}489 elif operator == "/":490 update_ops = {}491 for update_column in update_columns:492 update_ops[update_column] = 0.5 # placeholder493 return {"$mul": update_ops}494 elif operator == "=":495 update_ops = {}496 for update_column in update_columns:497 update_ops[update_column] = 1 # placeholder498 return {"$set": update_ops}499 else:500 raise NotImplementedError501 def _update_update_op(self, row, update_op, update_columns):502 """ Update the operator for a single row503 :param row: a new row504 :type row: Dict[str, object]505 :param update_op: an operator that returned by `get_update_op`506 :type update_op: Dict[str, Dict[str, float]]507 :param update_columns: the columns to update508 :type update_columns: List[str]509 :return: Dict[str, Dict[str, float]]510 :rtype: Dict[str, Dict[str, float]]511 """512 new_update_op = update_op.copy()513 for k, v in new_update_op.items():514 if k == "$inc":515 for update_column in update_columns:516 if v[update_column] == 1:517 v[update_column] = row[update_column]518 else:519 v[update_column] = -row[update_column]520 elif k == "$mul":521 for update_column in update_columns:522 if v[update_column] == 2:523 v[update_column] = row[update_column]524 else:525 v[update_column] = 1.0 / row[update_column]526 elif k == "$set":527 for update_column in update_columns:528 v[update_column] = row[update_column]529 return new_update_op530 def update_row(self, table_name, row, update_op, update_columns):531 """ Update a row that exists in a table532 (suggestion: consider to use `update_rows` if you want to update multiple rows)533 :param table_name: the table name to update534 :type table_name: str535 :param row: a new row536 :type row: Dict[str, object]537 :param update_op: an operator that returned by `get_update_op`538 :type update_op: Dict[str, Dict[str, float]]539 :param update_columns: the columns to update540 :type update_columns: List[str]541 """542 self._conn[table_name].update_one({"_id": row["_id"]}, self._update_update_op(row, update_op, update_columns))543 def update_rows(self, table_name, rows, update_ops, update_columns):544 """ Update rows that exist in a table545 :param table_name: the table name to update546 :type table_name: str547 :param rows: new rows548 :type rows: List[Dict[str, object]]549 :param update_ops: operator(s) that returned by `get_update_op`550 :type update_ops: Union[List[Dict[str, Dict[str, float]]], Dict[str, Dict[str, float]]]551 :param update_columns: the columns to update552 :type update_columns: List[str]553 """554 if len(rows) > 0:555 if isinstance(update_ops, (tuple, list)): # +-*/556 assert len(rows) == len(update_ops)557 update_op_collections = defaultdict(list)558 for i, row in enumerate(rows):559 # self.update_row(row, table_name, update_ops[i], update_columns)560 new_update_op = self._update_update_op(row, update_ops[i], update_columns)561 update_op_collections[json.dumps(new_update_op)].append(row)562 for new_update_op, op_rows in update_op_collections.items():563 new_update_op = json.loads(new_update_op)564 _ids = [row["_id"] for row in op_rows]565 for idx in range(0, len(_ids), self.chunksize):566 query = {"_id": {'$in': _ids[idx:idx + self.chunksize]}}567 self._conn[table_name].update_many(query, new_update_op)568 else: # =569 update_op = update_ops570 value_collections = defaultdict(list)571 for row in rows:572 value_collections[json.dumps([row[k] for k in update_columns])].append(row)573 for new_update_op, op_rows in value_collections.items():574 new_update_op = self._update_update_op(op_rows[0], update_op, update_columns)575 _ids = [row["_id"] for row in op_rows]576 for idx in range(0, len(_ids), self.chunksize):577 query = {"_id": {'$in': _ids[idx:idx + self.chunksize]}}578 self._conn[table_name].update_many(query, new_update_op)579 def get_rows_by_keys(self, table_name, bys, keys, columns, order_bys=None, reverse=False, top_n=None):580 """ Retrieve rows by specific keys in some order581 :param table_name: the table name to retrieve582 :type table_name: str583 :param bys: the given columns to match584 :type bys: List[str]585 :param keys: the given values to match586 :type keys: List[str]587 :param columns: the given columns to retrieve588 :type columns: List[str]589 :param order_bys: the columns whose value are used to sort rows590 :type order_bys: List[str]591 :param reverse: whether to sort in a reversed order592 :type reverse: bool593 :param top_n: how many rows to return, default `None` for all rows594 :type top_n: int595 :return: retrieved rows596 :rtype: List[Dict[str, object]]597 """598 query = OrderedDict(zip(bys, keys))599 projection = self.__get_projection(columns)600 cursor = self._conn[table_name].find(query, projection)601 if order_bys:602 direction = -1 if reverse else 1603 cursor = cursor.sort([(k, direction) for k in order_bys])604 if top_n:605 result = []606 for x in cursor:607 result.append(x)608 if len(result) >= top_n:609 break610 return result611 else:...
base.py
Source:base.py
1try:2 import ujson as json3except:4 import json5from collections import defaultdict, OrderedDict6class BaseConnection(object):7 def __init__(self, db_path, chunksize):8 self._conn = None9 self.chunksize = chunksize10 def close(self):11 if self._conn:12 self._conn.close()13 def __del__(self):14 self.close()15 def create_table(self, table_name, columns):16 raise NotImplementedError17 def get_columns(self, table_name, columns):18 raise NotImplementedError19 def select_row(self, table_name, _id, columns):20 raise NotImplementedError21 def select_rows(self, table_name, _ids, columns):22 raise NotImplementedError23 def insert_row(self, table_name, row):24 raise NotImplementedError25 def insert_rows(self, table_name, rows):26 raise NotImplementedError27 def update_row(self, table_name, row, update_op, update_columns):28 raise NotImplementedError29 def update_rows(self, table_name, rows, update_ops, update_columns):30 raise NotImplementedError31 def get_update_op(self, update_columns, operator):32 raise NotImplementedError33 def get_rows_by_keys(self, table_name, bys, keys, columns, order_bys=None, reverse=False, top_n=None):34 raise NotImplementedError35class SqliteConnection(BaseConnection):36 def __init__(self, db_path, chunksize):37 import sqlite338 super(SqliteConnection, self).__init__(db_path, chunksize)39 self._conn = sqlite3.connect(db_path)40 def create_table(self, table_name, columns, columns_types):41 create_table = "CREATE TABLE %s (%s);" % (table_name, ",".join(42 [' '.join(x) for x in zip(columns, columns_types)]))43 self._conn.execute(create_table)44 self._conn.commit()45 def get_columns(self, table_name, columns):46 select_table = "SELECT %s FROM %s;" % (",".join(columns), table_name)47 result = list(map(lambda x: OrderedDict(zip(columns, x)), self._conn.execute(select_table)))48 return result49 def select_row(self, table_name, _id, columns):50 select_table = "SELECT %s FROM %s WHERE _id=?;" % (",".join(columns), table_name)51 result = list(self._conn.execute(select_table, [_id]))52 if len(result) == 0:53 return None54 else:55 return OrderedDict(zip(columns, result[0]))56 def select_rows(self, table_name, _ids, columns):57 if len(_ids) > 0:58 row_cache = dict()59 result = []60 for idx in range(0, len(_ids), self.chunksize):61 select_table = "SELECT %s FROM %s WHERE _id IN ('%s');" % (62 ",".join(columns), table_name, "','".join(_ids[idx:idx+self.chunksize]))63 result.extend(list(self._conn.execute(select_table)))64 for x in result:65 exact_match_row = OrderedDict(zip(columns, x))66 row_cache[exact_match_row["_id"]] = exact_match_row67 exact_match_rows = []68 for _id in _ids:69 exact_match_rows.append(row_cache.get(_id, None))70 return exact_match_rows71 else:72 return []73 def insert_row(self, table_name, row):74 insert_table = "INSERT INTO %s VALUES (%s)" % (table_name, ",".join(['?'] * (len(row))))75 self._conn.execute(insert_table, list(row.values()))76 self._conn.commit()77 def insert_rows(self, table_name, rows):78 if len(rows) > 0:79 insert_table = "INSERT INTO %s VALUES (%s)" % (table_name, ",".join(['?'] * (len(next(iter(rows))))))80 self._conn.executemany(insert_table, [list(row.values()) for row in rows])81 self._conn.commit()82 def _update_update_op(self, row, update_op, update_columns):83 update_op_sp = update_op.split('?')84 while len(update_op_sp) >= 0 and update_op_sp[-1] == '':85 update_op_sp.pop()86 assert len(update_op_sp) == len(update_columns)87 new_update_op = []88 for i in range(len(update_op_sp)):89 new_update_op.append(update_op_sp[i])90 if isinstance(row[update_columns[i]], str):91 new_update_op.append("'" + row[update_columns[i]].replace("'", "''") + "'")92 else:93 new_update_op.append(str(row[update_columns[i]]))94 return ''.join(new_update_op)95 def update_row(self, table_name, row, update_op, update_columns):96 update_table = "UPDATE %s SET %s WHERE _id=?" % (table_name, update_op)97 self._conn.execute(update_table, [row[k] for k in update_columns] + [row["_id"]])98 self._conn.commit()99 def update_rows(self, table_name, rows, update_ops, update_columns):100 if len(rows) > 0:101 if isinstance(update_ops, (tuple, list)): # +-*/102 assert len(rows) == len(update_ops)103 # group rows by op to speed up104 update_op_collections = defaultdict(list) # key: _update_update_op105 for i, row in enumerate(rows):106 # self.update_row(row, table_name, update_ops[i], update_columns)107 new_update_op = self._update_update_op(row, update_ops[i], update_columns)108 update_op_collections[new_update_op].append(row)109 for new_update_op, op_rows in update_op_collections.items():110 _ids = [row["_id"] for row in op_rows]111 for idx in range(0, len(_ids), self.chunksize):112 update_table = "UPDATE %s SET %s WHERE _id IN ('%s');" % (113 table_name, new_update_op, "','".join(_ids[idx:idx+self.chunksize]))114 self._conn.execute(update_table)115 else: # =116 update_op = update_ops117 # group rows by new values to speed up118 value_collections = defaultdict(list) # key: values of new values119 for row in rows:120 # self.update_row(row, table_name, update_op, update_columns)121 value_collections[json.dumps([row[k] for k in update_columns])].append(row)122 for new_update_op, op_rows in value_collections.items():123 new_update_op = self._update_update_op(op_rows[0], update_op, update_columns)124 _ids = [row["_id"] for row in op_rows]125 for idx in range(0, len(_ids), self.chunksize):126 update_table = "UPDATE %s SET %s WHERE _id IN ('%s');" % (127 table_name, new_update_op, "','".join(_ids[idx:idx+self.chunksize]))128 self._conn.execute(update_table)129 self._conn.commit()130 """131 if isinstance(update_ops, list) or isinstance(update_ops, tuple):132 assert len(rows) == len(update_ops)133 for i, row in enumerate(rows):134 self.update_row(row, table_name, update_ops[i], update_columns)135 else:136 update_op = update_ops137 update_table = "UPDATE %s SET %s WHERE _id=?" % (138 table_name, update_op)139 self._conn.executemany(140 update_table, [[row[k] for k in update_columns] + [row["_id"]] for row in rows])141 self._conn.commit()142 """143 def get_update_op(self, update_columns, operator):144 if operator in "+-*/":145 update_ops = []146 for update_column in update_columns:147 update_ops.append(update_column + "=" + update_column + operator + "?")148 return ",".join(update_ops)149 elif operator == "=":150 update_ops = []151 for update_column in update_columns:152 update_ops.append(update_column + "=?")153 return ",".join(update_ops)154 else:155 raise NotImplementedError156 def get_rows_by_keys(self, table_name, bys, keys, columns, order_bys=None, reverse=False, top_n=None):157 key_match_events = []158 select_table = "SELECT %s FROM %s WHERE %s" % (159 ",".join(columns), table_name, " AND ".join(["%s=?" % (by) for by in bys]))160 if order_bys:161 select_table += " ORDER BY %s %s" % (",".join(order_bys), "DESC" if reverse else "ASC")162 if top_n:163 select_table += " LIMIT %d" % (top_n)164 select_table += ";"165 for x in self._conn.execute(select_table, keys):166 key_match_event = OrderedDict(zip(columns, x))167 key_match_events.append(key_match_event)168 return key_match_events169class MongoDBConnection(BaseConnection):170 def __init__(self, db_path, chunksize):171 import pymongo172 super(MongoDBConnection, self).__init__(db_path, chunksize)173 self._client = pymongo.MongoClient("mongodb://localhost:27017/", document_class=OrderedDict)174 self._conn = self._client[os.path.splitext(os.path.basename(db_path))[0]]175 def close(self):176 self._client.close()177 def create_table(self, table_name, columns, columns_types):178 self._conn[table_name]179 def __get_projection(self, columns):180 projection = {"_id": 0}181 for k in columns:182 projection[k] = 1183 return projection184 def get_columns(self, table_name, columns):185 projection = self.__get_projection(columns)186 results = list(self._conn[table_name].find({}, projection))187 return results188 def select_row(self, table_name, _id, columns):189 projection = self.__get_projection(columns)190 return self._conn[table_name].find_one({"_id": _id}, projection)191 def select_rows(self, table_name, _ids, columns):192 table = self._conn[table_name]193 exact_match_rows = []194 projection = self.__get_projection(columns)195 for idx in range(0, len(_ids), self.chunksize):196 query = {"_id": {'$in': _ids[idx:idx+self.chunksize]}}197 exact_match_rows.extend(table.find(query, projection))198 row_cache = {x["_id"]: x for x in exact_match_rows}199 exact_match_rows = [row_cache.get(_id, None) for _id in _ids]200 return exact_match_rows201 def insert_row(self, table_name, row):202 self._conn[table_name].insert_one(row)203 def insert_rows(self, table_name, rows):204 self._conn[table_name].insert_many(rows)205 def _update_update_op(self, row, update_op, update_columns):206 new_update_op = update_op.copy()207 for k, v in new_update_op.items():208 if k == "$inc":209 for update_column in update_columns:210 if v[update_column] == 1:211 v[update_column] = row[update_column]212 else:213 v[update_column] = -row[update_column]214 elif k == "$mul":215 for update_column in update_columns:216 if v[update_column] == 2:217 v[update_column] = row[update_column]218 else:219 v[update_column] = 1.0 / row[update_column]220 elif k == "$set":221 for update_column in update_columns:222 v[update_column] = row[update_column]223 return new_update_op224 def update_row(self, table_name, row, update_op, update_columns):225 self._conn[table_name].update_one(226 {"_id": row["_id"]}, self._update_update_op(row, update_op, update_columns))227 def update_rows(self, table_name, rows, update_ops, update_columns):228 if len(rows) > 0:229 if isinstance(update_ops, (tuple, list)): # +-*/230 assert len(rows) == len(update_ops)231 update_op_collections = defaultdict(list)232 for i, row in enumerate(rows):233 # self.update_row(row, table_name, update_ops[i], update_columns)234 new_update_op = self._update_update_op(row, update_ops[i], update_columns)235 update_op_collections[json.dumps(new_update_op)].append(row)236 for new_update_op, op_rows in update_op_collections.items():237 new_update_op = json.loads(new_update_op)238 _ids = [row["_id"] for row in op_rows]239 for idx in range(0, len(_ids), self.chunksize):240 query = {"_id": {'$in': _ids[idx:idx+self.chunksize]}}241 self._conn[table_name].update_many(query, new_update_op)242 else: # =243 update_op = update_ops244 value_collections = defaultdict(list)245 for row in rows:246 # self.update_row(row, table_name, update_op, update_columns)247 value_collections[json.dumps([row[k] for k in update_columns])].append(row)248 for new_update_op, op_rows in value_collections.items():249 new_update_op = self._update_update_op(op_rows[0], update_op, update_columns)250 _ids = [row["_id"] for row in op_rows]251 for idx in range(0, len(_ids), self.chunksize):252 query = {"_id": {'$in': _ids[idx:idx+self.chunksize]}}253 self._conn[table_name].update_many(query, new_update_op)254 def get_update_op(self, update_columns, operator):255 if operator == "+":256 update_ops = {}257 for update_column in update_columns:258 update_ops[update_column] = 1 # placeholder259 return {"$inc": update_ops}260 elif operator == "-":261 update_ops = {}262 for update_column in update_columns:263 update_ops[update_column] = -1 # placeholder264 return {"$inc": update_ops}265 elif operator == "*":266 update_ops = {}267 for update_column in update_columns:268 update_ops[update_column] = 2 # placeholder269 return {"$mul": update_ops}270 elif operator == "/":271 update_ops = {}272 for update_column in update_columns:273 update_ops[update_column] = 0.5 # placeholder274 return {"$mul": update_ops}275 elif operator == "=":276 update_ops = {}277 for update_column in update_columns:278 update_ops[update_column] = 1 # placeholder279 return {"$set": update_ops}280 else:281 raise NotImplementedError282 def get_rows_by_keys(self, table_name, bys, keys, columns, order_bys=None, reverse=False, top_n=None):283 query = OrderedDict(zip(bys, keys))284 projection = self.__get_projection(columns)285 cursor = self._conn[table_name].find(query, projection)286 if order_bys:287 direction = -1 if reverse else 1288 cursor = cursor.sort([(k, direction) for k in order_bys])289 if top_n:290 result = []291 for x in cursor:292 result.append(x)293 if len(result) >= top_n:294 break295 return result296 else:...
dnn_ncrt_liew_combined_latest.py
Source:dnn_ncrt_liew_combined_latest.py
...39 predicted = list(np.round(predicted))40 predicted = list(map(lambda row: list(map(int, row)), predicted))41 return predicted, classifier42def get_fingerprints(mol):43 morgan = update_columns(morgan_fingerprints(mol), 'morgan')44 maccs = update_columns(maccs_fingerprints(mol), 'maccs')45 charge_f, e = charge_features(mol)46 autocorrelation_f, e = autocorrelation_features(mol)47 harmonic_topology = pd.DataFrame(harmonic_topology_index_feature(mol))48 constitutional_f = constitutional_features(mol)49 estate_f = estate_features(mol)50 moe_f = moe_features(mol)51 bcut_f, e = bcut_features(mol)52 molproperty_f, e = molproperty_features(mol)53 cats2d_f = cats2d_features(mol)54 kappa = kappa_descriptors(mol)55 return {56 "maccs": maccs,57 "morgan": morgan,58 'charge': update_columns(charge_f, 'charge'),59 'harmonic_topology': update_columns(harmonic_topology, 'harmonic'),60 'autocorrelation': update_columns(autocorrelation_f, 'autocorrelation'),61 'constitutional': update_columns(constitutional_f, 'constitutional'),62 'estate': update_columns(estate_f, 'estate'),63 'moe': update_columns(moe_f, 'moe'),64 'bcut': update_columns(bcut_f, 'bcut'),65 'molproperty': update_columns(molproperty_f, 'molproperty'),66 'cats2d': update_columns(cats2d_f, 'cats2d'),67 'kappa': update_columns(kappa, 'kappa'),68 }69if __name__ == "__main__":70 ncrt_train = pd.read_csv('./data/raw/ncrt_liew_train.csv', index_col=0).fillna(0).reset_index(drop=True)71 ncrt_test = pd.read_csv('./data/raw/ncrt_liew_test.csv', index_col=0).fillna(0).reset_index(drop=True)72 common_comb = ['maccs', 'morgan', 'charge', 'harmonic_topology', 'autocorrelation', 'constitutional', 'estate',73 'moe', 'bcut', 'molproperty', 'cats2d', 'kappa']74 fingerprints = get_fingerprints(convert_to_mol(ncrt_train['smiles']))75 fingerprints_test = get_fingerprints(convert_to_mol(ncrt_test['smiles']))76 all_inputs = pd.concat(list(map(lambda x: fingerprints[x], common_comb)), axis=1).fillna(0)77 all_inputs_test = pd.concat(list(map(lambda x: fingerprints_test[x], common_comb)), axis=1).fillna(0)78 X = pd.concat([all_inputs, all_inputs_test]).reset_index(drop=True)79 y = pd.concat([ncrt_train['label'], ncrt_test['label']]).reset_index(drop=True)80 # Instantiate the cross validator81 skf = StratifiedKFold(n_splits=10, random_state=1)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pandera automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
