spacepaste

Paste details

reply | raw

  1.  
  2. @async_generator
  3. async def execute_select_query(engine, query_string, fetch_size, flush_after=None):
  4. async with engine.connect() as conn:
  5. try:
  6. result = await conn.execute(query_string)
  7. empty = False
  8. while not empty:
  9. batch = await result.fetchmany(fetch_size)
  10. if not batch:
  11. empty = True
  12. for row in batch:
  13. await yield_(row)
  14. finally:
  15. if flush_after is not None:
  16. await conn.execute(f"FLUSH TABLE `{flush_after}`")
  18. async def primary_key_finder(engine, table_name):
  19. async with engine.connect() as conn:
  20. get_key_query = await conn.execute(
  21. f"SHOW KEYS FROM `{table_name}` WHERE Key_name = %s", ("PRIMARY",)
  22. )
  23. await conn.execute(f"FLUSH TABLE `{table_name}`")
  24. keys = await get_key_query.fetchall()
  25. return [row["Column_name"] for row in keys]
  28. async def check_for_mojibake(row, found_mojibake, database, table, has_pri_key=False):
  29. for column, data in row.items():
  30. if data is None or not column.startswith("HEX"):
  31. continue
  32. field_bytes = binascii.unhexlify(data)
  33. try:
  34. field_bytes.decode("ascii")
  35. except UnicodeDecodeError:
  36. try:
  37. field_bytes.decode("utf-8")
  38. if not has_pri_key:
  39. log_keyless_table(database, table)
  40. raise ValueError
  41. found_mojibake[database][table][rm_hex_wrap(column)]["rows_found"] += 1
  42. except UnicodeDecodeError:
  43. pass
  46. async def table_mojibake_scan(
  47. engine, database, table, table_columns, found_mojibake, table_limiter
  48. ):
  49. hex_wrapped_cols = ", ".join(
  50. "HEX(`{0}`)".format(column) for column in table_columns["text_columns"]
  51. )
  52. has_pri_key = False
  53. pri_key = await primary_key_finder(engine, table)
  54. if pri_key:
  55. has_pri_key = True
  56. key_cols = ", ".join("`{0}`".format(key_col) for key_col in pri_key)
  57. cols_query = f"SELECT {hex_wrapped_cols}, {key_cols} FROM `{table}`"
  58. else:
  59. cols_query = f"SELECT {hex_wrapped_cols} FROM `{table}`"
  60. with warnings.catch_warnings():
  61. warnings.simplefilter("ignore")
  62. async with aclosing(
  63. execute_select_query(engine, cols_query, fetch_size=500, flush_after=table)
  64. ) as text_columns:
  65. async for row in text_columns:
  66. try:
  67. await check_for_mojibake(
  68. row, found_mojibake, database, table, has_pri_key
  69. )
  70. except ValueError:
  71. break
  72. table_limiter.release()
  73.  
Powered by spacepaste. Paste removal.