批量删除 MySQL 表

mysql_drop_tables.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

u"""MySQL Drop Tables

Usage:
  mysql_drop_tables.py [-y] MYSQL_URL TABLE_PATTERN ...

Options:
  -y                直接执行，不提示用户
"""

from ast import literal_eval
from docopt import docopt
import pymysql
import sys
import time
import urlparse

VERSION = 'MySQL Drop Tables 0.1'

urlparse.uses_netloc.append('mysql')
urlparse.uses_query.append('mysql')

def setDefaultEncoding():
    reload(sys)
    if sys.platform == 'win32':
        sys.setdefaultencoding('cp936')
    else:
        sys.setdefaultencoding('utf-8')

setDefaultEncoding()

def parseValue(value):
    try:
        return literal_eval(value)
    except:
        pass
    try:
        return literal_eval('"%s"' % value)
    except:
        pass
    return value


def parseMySQLUrl(url):
    url = urlparse.urlparse(url)
    database = url.path.strip('/')

    config = {
        'user': url.username,
        'passwd': url.password,
        'host': url.hostname or 'localhost',
        'port': url.port or 3306,
        'database': database,
        }

    query = urlparse.parse_qs(url.query)
    for name, value in query.iteritems():
        config[name] = parseValue(value[0])

    return config

def getAnswer(prompt):
    while True:
        answer = raw_input(prompt).lower()
        if answer and answer[0] in 'yna':
            return answer

class Operation(object):
    def __init__(self, mysqlUrl, tablePatterns, yes):
        self.mysqlUrl = mysqlUrl
        self.tablePatterns = tablePatterns
        self.yes = yes

    def connect(self):
        config = parseMySQLUrl(self.mysqlUrl)
    
        try:
            self.db = pymysql.connect(**config)
        except pymysql.DatabaseError as e:
            print e, config

    def getTableNames(self):
        for pattern in self.tablePatterns:
            print '-- 删除 %r --' % pattern
            try:
                cursor = self.db.cursor()
                cursor.execute('SHOW TABLES LIKE %s', pattern)
                tableNames = cursor.fetchall()
                cursor.close()
            except pymysql.DatabaseError as e:
                print e
            else:
                for tableName in tableNames:
                    yield tableName

    def dropTable(self, tableName):
        print u'删除 %r ...' % tableName,
        startTime = time.time()
        try:
            cursor = self.db.cursor()
            cursor.execute('DROP TABLE %s' % tableName)
            cursor.close()
        except pymysql.DatabaseError as e:
            print e
        ms = int((time.time() - startTime) * 1000)
        print u'耗时 %d ms' % ms

    def process(self):
        for tableName in self.getTableNames():
            if not self.yes:
                answer = getAnswer(u'确实要删除 %r 吗(Yes/No/All)?' % tableName)
                if answer == 'n':
                    continue
                if answer == 'a':
                    self.yes = True
            self.dropTable(tableName)

    def close(self):
        self.db.close()


def main():
    arguments = docopt(__doc__, version=VERSION)

    mysqlUrl = arguments.get('MYSQL_URL')
    tablePatterns = arguments.get('TABLE_PATTERN')
    yes = arguments.get('-y')

    op = Operation(mysqlUrl, tablePatterns, yes)
    op.connect()
    op.process()
    op.close()


if __name__ == '__main__':
    main()

配合使用的脚本文件

!/bin/sh

echo 开始时间 $(date '+%F %T') >> /tmp/mydb-clean.log
/home/wts/tools/mysql_drop_tables.py -y mysql://root@localhost/mydb $(date --date '105 days ago' '+%%%Y%m%d') >> /tmp/mydb-clean.log
echo 结束时间 $(date '+%F %T') >> /tmp/mydb-clean.log