1import hashlib 

2 

3from .utils import ( 

4 detect_spatialite, 

5 detect_fts, 

6 detect_primary_keys, 

7 escape_sqlite, 

8 get_all_foreign_keys, 

9 table_columns, 

10 sqlite3, 

11) 

12 

13 

14HASH_BLOCK_SIZE = 1024 * 1024 

15 

16 

17def inspect_hash(path): 

18 " Calculate the hash of a database, efficiently. " 

19 m = hashlib.sha256() 

20 with path.open("rb") as fp: 

21 while True: 

22 data = fp.read(HASH_BLOCK_SIZE) 

23 if not data: 

24 break 

25 m.update(data) 

26 

27 return m.hexdigest() 

28 

29 

30def inspect_views(conn): 

31 " List views in a database. " 

32 return [ 

33 v[0] for v in conn.execute('select name from sqlite_master where type = "view"') 

34 ] 

35 

36 

37def inspect_tables(conn, database_metadata): 

38 " List tables and their row counts, excluding uninteresting tables. " 

39 tables = {} 

40 table_names = [ 

41 r["name"] 

42 for r in conn.execute('select * from sqlite_master where type="table"') 

43 ] 

44 

45 for table in table_names: 

46 table_metadata = database_metadata.get("tables", {}).get(table, {}) 

47 

48 try: 

49 count = conn.execute( 

50 "select count(*) from {}".format(escape_sqlite(table)) 

51 ).fetchone()[0] 

52 except sqlite3.OperationalError: 

53 # This can happen when running against a FTS virtual table 

54 # e.g. "select count(*) from some_fts;" 

55 count = 0 

56 

57 column_names = table_columns(conn, table) 

58 

59 tables[table] = { 

60 "name": table, 

61 "columns": column_names, 

62 "primary_keys": detect_primary_keys(conn, table), 

63 "count": count, 

64 "hidden": table_metadata.get("hidden") or False, 

65 "fts_table": detect_fts(conn, table), 

66 } 

67 

68 foreign_keys = get_all_foreign_keys(conn) 

69 for table, info in foreign_keys.items(): 

70 tables[table]["foreign_keys"] = info 

71 

72 # Mark tables 'hidden' if they relate to FTS virtual tables 

73 hidden_tables = [ 

74 r["name"] 

75 for r in conn.execute( 

76 """ 

77 select name from sqlite_master 

78 where rootpage = 0 

79 and sql like '%VIRTUAL TABLE%USING FTS%' 

80 """ 

81 ) 

82 ] 

83 

84 if detect_spatialite(conn): 

85 # Also hide Spatialite internal tables 

86 hidden_tables += [ 

87 "ElementaryGeometries", 

88 "SpatialIndex", 

89 "geometry_columns", 

90 "spatial_ref_sys", 

91 "spatialite_history", 

92 "sql_statements_log", 

93 "sqlite_sequence", 

94 "views_geometry_columns", 

95 "virts_geometry_columns", 

96 ] + [ 

97 r["name"] 

98 for r in conn.execute( 

99 """ 

100 select name from sqlite_master 

101 where name like "idx_%" 

102 and type = "table" 

103 """ 

104 ) 

105 ] 

106 

107 for t in tables.keys(): 

108 for hidden_table in hidden_tables: 

109 if t == hidden_table or t.startswith(hidden_table): 

110 tables[t]["hidden"] = True 

111 continue 

112 

113 return tables