"""Entrypoint.""" import os import pymysql user = os.environ['DB_USER'] password = os.environ['DB_PASSWORD'] host = os.environ['DB_HOST'] schema = os.environ['DB_SCHEMA'] port = os.environ.get('DB_PORT', 3306) DB_CONN = pymysql.connect( host=host, user=user, password=password, database=schema, cursorclass=pymysql.cursors.DictCursor ) INT_MAP = { 'TINYINT': (127, 255), 'SMALLINT': (32767, 65535), 'MEDIUMINT': (8388607, 16777215), 'INT': (2147483647, 4294967295), 'BIGINT': (9223372036854775807, 18446744073709551615) } def main(): """Docstring.""" _valid(schema) table_names = [ x[f'Tables_in_{schema}'] for x in query(f'SHOW FULL tables FROM {schema} WHERE Table_type = "BASE TABLE";') # noqa:E501 ] for table in table_names: results = query(f'SHOW CREATE TABLE `{table}`;') create_stmt = results[0]['Create Table'] lines = [ y for y in create_stmt.split('\n') if 'AUTO_INCREMENT' in y ] if len(lines) != 2: continue int_type = lines[0].split()[1].upper() unsigned = lines[0].split()[2].upper() == 'UNSIGNED' int_value = int([x for x in lines[1].split() if 'AUTO_INCREMENT' in x][0].split('=')[1]) # noqa:E501 max_int = INT_MAP[int_type][1] if unsigned else INT_MAP[int_type][0] # noqa:E501 percent_full = (int_value / max_int) * 100 print(f'{table},{int_type},{unsigned},{percent_full}') def query(sql, params=dict()): """Perform query.""" with DB_CONN.cursor() as cursor: cursor.execute(sql, params) results = cursor.fetchall() return results def _valid(string): if not string.replace('_', '').isalnum(): raise Exception(f'{string} not valid for direct query injection') return string if __name__ == '__main__': main()