欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

利用Python批量導(dǎo)出mysql數(shù)據(jù)庫(kù)表結(jié)構(gòu)的操作實(shí)例

 更新時(shí)間:2022年08月08日 14:56:42   作者:wdbrmeng  
這篇文章主要給大家介紹了關(guān)于利用Python批量導(dǎo)出mysql數(shù)據(jù)庫(kù)表結(jié)構(gòu)的相關(guān)資料,需要的朋友可以參考下

前言

最近在公司售前售后同事遇到一些奇怪的需求找到我,需要提供公司一些項(xiàng)目數(shù)據(jù)庫(kù)所有表的結(jié)構(gòu)信息(字段名、類(lèi)型、長(zhǎng)度、是否主鍵、***、備注),雖然不是本職工作,但是作為python技能的擁有者看到這種需求還是覺(jué)得很容易的,但是如果不用代碼解決確實(shí)非常棘手和浪費(fèi)時(shí)間。于是寫(xiě)了一個(gè)輕量小型項(xiàng)目來(lái)解決一些燃眉之急,希望能對(duì)一些人有所幫助,代碼大神、小神可以忽略此貼。

代碼直達(dá): GITEE、GitHub

解決方法

1. mysql 數(shù)據(jù)庫(kù) 表信息查詢(xún)

想要導(dǎo)出mysql數(shù)據(jù)庫(kù)表結(jié)構(gòu)必須了解一些相關(guān)數(shù)據(jù)庫(kù)知識(shí),mysql數(shù)據(jù)庫(kù)支持通過(guò)SQL語(yǔ)句進(jìn)行表信息查詢(xún):

查詢(xún)數(shù)據(jù)庫(kù)所有表名

SHOW TABLES

查詢(xún)對(duì)應(yīng)數(shù)據(jù)庫(kù)對(duì)應(yīng)表結(jié)構(gòu)信息

SELECT COLUMN_NAME,COLUMN_TYPE,COLUMN_KEY,IS_NULLABLE, COLUMN_COMMENT 
FROM information_schema.`COLUMNS` 
WHERE TABLE_SCHEMA='{dbName}' AND TABLE_NAME='{tableName}'
  • COLUMN_NAME:字段名
  • COLUMN_TYPE:數(shù)據(jù)類(lèi)型
  • COLUMN_KEY:主鍵
  • IS_NULLABLE:非空
  • COLUMN_COMMENT:字段描述
    還有一些其他字段,有需要可自行百度

2.連接數(shù)據(jù)庫(kù)代碼

以下是一個(gè)較為通用的mysql數(shù)據(jù)庫(kù)連接類(lèi),創(chuàng)建 MysqlConnection 類(lèi),放入對(duì)應(yīng)數(shù)據(jù)庫(kù)連接信息即可使用sql,通過(guò)query查詢(xún)、update增刪改、close關(guān)閉連接。

*注:數(shù)據(jù)量過(guò)大時(shí)不推薦直接使用query查詢(xún)。

import pymysql

class MysqlConnection():
    def __init__(self, host, user, passw, port, database, charset="utf8"):
        self.db = pymysql.connect(host=host, user=user, password=passw, port=port,
                                  database=database, charset=charset)
        self.cursor = self.db.cursor()

    # 查
    def query(self, sql):
        self.cursor.execute(sql)
        results = self.cursor.fetchall()
        return results

    # 增刪改
    def update(self, sql):
        try:
            self.cursor.execute(sql)
            self.db.commit()
            return 1
        except Exception as e:
            print(e)
            self.db.rollback()
            return 0

    # 關(guān)閉連接
    def close(self):
        self.cursor.close()
        self.db.close()

3.數(shù)據(jù)查詢(xún)處理代碼

3.0 配置信息

config.yml,這里使用了配置文件進(jìn)行程序參數(shù)配置,方便配置一鍵運(yùn)行

# 數(shù)據(jù)庫(kù)信息配置
db_config:
  host: 127.0.0.1	# 數(shù)據(jù)庫(kù)所在服務(wù)IP
  port: 3306		# 數(shù)據(jù)庫(kù)服務(wù)端口
  username: root	# ~用戶(hù)名
  password: 12346	# ~密碼
  charset: utf8
  # 需要進(jìn)行處理的數(shù)據(jù)名稱(chēng)列表 《《 填入數(shù)據(jù)庫(kù)名
  db_names: ['db_a','db_b']

# 導(dǎo)出配置
excel_conf:
  # 導(dǎo)出結(jié)構(gòu)Excel表頭,長(zhǎng)度及順序不可調(diào)整,僅支持更換名稱(chēng)
  column_name: ['字段名', '數(shù)據(jù)類(lèi)型', '長(zhǎng)度', '主鍵', '非空', '描述']
  save_dir: ./data

讀取配置文件的代碼

import yaml

class Configure():
    def __init__(self):
        with open("config.yaml", 'r', encoding='utf-8') as f:
            self._conf = yaml.load(f.read(), Loader=yaml.FullLoader)

    def get_db_config(self):
        host = self._conf['db_config']['host']
        port = self._conf['db_config']['port']
        username = self._conf['db_config']['username']
        password = self._conf['db_config']['password']
        charset = self._conf['db_config']['charset']
        db_names = self._conf['db_config']['db_names']
        return host, port, username, password, charset, db_names

    def get_excel_title(self):
        title = self._conf['excel_conf']['column_name']
        save_dir = self._conf['excel_conf']['save_dir']
        return title, save_dir

3.1查詢(xún)數(shù)據(jù)庫(kù)表

利用上面創(chuàng)建的數(shù)據(jù)庫(kù)連接和SQL查詢(xún)獲取所有表

class ExportMysqlTableStructureInfoToExcel():
	def __init__(self):
	        conf = Configure()	# 獲取配置初始化類(lèi)信息
	        self.__host, self.__port, self.__username, self.__password, self.__charset, self.db_names = conf.get_db_config()
	        self.__excel_title, self.__save_dir = conf.get_excel_title()
	```省略```
	def __connect_to_mysql(self, database):	# 獲取數(shù)據(jù)庫(kù)連接方法
        connect = MysqlConnection(self.__host,
                                  self.__username,
                                  self.__password,
                                  self.__port, database,
                                  self.__charset)
        return connect
        
	def __get_all_tables(self, con):	# 查詢(xún)所有表
	        res = con.query("SHOW TABLES")
	        tb_list = []
	        for item in res:
	            tb_list.append(item[0])
	        return tb_list
	``````

3.2 查詢(xún)對(duì)應(yīng)表結(jié)構(gòu)

循環(huán)獲取每一張表的結(jié)構(gòu)數(shù)據(jù),根據(jù)需要對(duì)中英文做了一些轉(zhuǎn)換,字段長(zhǎng)度可以從類(lèi)型中分離出來(lái),這里使用yield返回?cái)?shù)據(jù),可以利用生成器加速處理過(guò)程(外包導(dǎo)出保存和數(shù)據(jù)庫(kù)查詢(xún)可以并行)

class ExportMysqlTableStructureInfoToExcel():
	```省略```
	def __struct_of_table_generator(self, con, db_name):
        tb_list = self.__get_all_tables(con)
        for index, tb_name in enumerate(tb_list):
            sql = "SELECT COLUMN_NAME,COLUMN_TYPE,COLUMN_KEY,IS_NULLABLE, COLUMN_COMMENT " \
              "FROM information_schema.`COLUMNS` WHERE TABLE_SCHEMA='{}' AND TABLE_NAME='{}'".format(db_name, tb_name)
            res = con.query(sql)
            struct_list = []
            for item in res:
                column_name, column_type, column_key, is_nullable, column_comment = item
                length = "0"
                if str(column_type).find('(') > -1:
                    column_type, length = str(column_type).replace(")", '').split('(')
                if column_key == 'PRI':
                    column_key = "是"
                else:
                    column_key = ''
                if is_nullable == 'YES':
                    is_nullable = '是'
                else:
                    is_nullable = '否'
                struct_list.append([column_name, column_type, length, column_key, is_nullable, column_comment])
            yield [struct_list, tb_name]
	```省略```

3.3 pandas進(jìn)行數(shù)據(jù)保存導(dǎo)出excel

class ExportMysqlTableStructureInfoToExcel():
	```省略```
	def export(self):
        if len(self.db_names) == 0:
            print("請(qǐng)配置數(shù)據(jù)庫(kù)列表")
        for i, db_name in enumerate(self.db_names):		# 對(duì)多個(gè)數(shù)據(jù)庫(kù)進(jìn)行處理
            connect = self.__connect_to_mysql(db_name)	# 獲取數(shù)據(jù)庫(kù)連接
            if not os.path.exists(self.__save_dir):		# 判斷數(shù)據(jù)導(dǎo)出保存路徑是否存在
                os.mkdir(self.__save_dir)

            file_name = os.path.join(self.__save_dir,'{}.xlsx'.format(db_name))	# 用數(shù)據(jù)庫(kù)名命名導(dǎo)出Excel文件
            if not os.path.exists(file_name):  # 文件不存在時(shí)自動(dòng)創(chuàng)建文件 excel
                wrokb = openpyxl.Workbook()
                wrokb.save(file_name)
                wrokb.close()
            wb = openpyxl.load_workbook(file_name)
            writer = pd.ExcelWriter(file_name, engine='openpyxl')
            writer.book = wb

            struct_generator = self.__struct_of_table_generator(connect, db_name)	# 獲取表結(jié)構(gòu)信息的生成器

            for tb_info in tqdm(struct_generator, desc=db_name):	# 從生成器中獲取表結(jié)構(gòu)并利用pandas進(jìn)行格式化保存,寫(xiě)入Excel文件
                s_list, tb_name = tb_info
                data = pd.DataFrame(s_list, columns=self.__excel_title)
                data.to_excel(writer, sheet_name=tb_name)
            writer.close()

            connect.close()
	```省略```

補(bǔ)充:python腳本快速生成mysql數(shù)據(jù)庫(kù)結(jié)構(gòu)文檔

由于數(shù)據(jù)表太多,手動(dòng)編寫(xiě)耗費(fèi)的時(shí)間太久,所以搞了一個(gè)簡(jiǎn)單的腳本快速生成數(shù)據(jù)庫(kù)結(jié)構(gòu),保存到word文檔中。

1.安裝pymysql和document

pip install pymysql
pip install document

2.腳本

# -*- coding: utf-8 -*-
import pymysql
from docx import Document
from docx.shared import Pt
from docx.oxml.ns import qn

db = pymysql.connect(host='127.0.0.1', #數(shù)據(jù)庫(kù)服務(wù)器IP
                         port=3306,
                         user='root',
                         passwd='123456',
                         db='test_db') #數(shù)據(jù)庫(kù)名稱(chēng))
#根據(jù)表名查詢(xún)對(duì)應(yīng)的字段相關(guān)信息
def query(tableName):
    #打開(kāi)數(shù)據(jù)庫(kù)連接
    cur = db.cursor()
    sql = "select b.COLUMN_NAME,b.COLUMN_TYPE,b.COLUMN_COMMENT from (select * from information_schema.`TABLES`  where TABLE_SCHEMA='test_db') a right join(select * from information_schema.`COLUMNS` where TABLE_SCHEMA='test_db_test') b on a.TABLE_NAME = b.TABLE_NAME where a.TABLE_NAME='" + tableName+"'"
    cur.execute(sql)
    data = cur.fetchall()
    cur.close
    return data
#查詢(xún)當(dāng)前庫(kù)下面所有的表名,表名:tableName;表名+注釋(用于填充至word文檔):concat(TABLE_NAME,'(',TABLE_COMMENT,')')
def queryTableName():
    cur = db.cursor()
    sql = "select TABLE_NAME,concat(TABLE_NAME,'(',TABLE_COMMENT,')') from information_schema.`TABLES`  where TABLE_SCHEMA='test_db_test'"
    cur.execute(sql)
    data = cur.fetchall()
    return data
#將每個(gè)表生成word結(jié)構(gòu),輸出到word文檔
def generateWord(singleTableData,document,tableName):
    p=document.add_paragraph()
    p.paragraph_format.line_spacing=1.5 #設(shè)置該段落 行間距為 1.5倍
    p.paragraph_format.space_after=Pt(0) #設(shè)置段落 段后 0 磅
    #document.add_paragraph(tableName,style='ListBullet')
    r=p.add_run('\n'+tableName)
    r.font.name=u'宋體'
    r.font.size=Pt(12)
    table = document.add_table(rows=len(singleTableData)+1, cols=3,style='Table Grid')
    table.style.font.size=Pt(11)
    table.style.font.name=u'Calibri'
    #設(shè)置表頭樣式
    #這里只生成了三個(gè)表頭,可通過(guò)實(shí)際需求進(jìn)行修改
    for i in ((0,'NAME'),(1,'TYPE'),(2,'COMMENT')):
        run = table.cell(0,i[0]).paragraphs[0].add_run(i[1])
        run.font.name = 'Calibri'
        run.font.size = Pt(11)
        r = run._element
        r.rPr.rFonts.set(qn('w:eastAsia'), '宋體')
    
    for i in range(len(singleTableData)):
        #設(shè)置表格內(nèi)數(shù)據(jù)的樣式
        for j in range(len(singleTableData[i])):
            run = table.cell(i+1,j).paragraphs[0].add_run(singleTableData[i][j])
            run.font.name = 'Calibri'
            run.font.size = Pt(11)
            r = run._element
            r.rPr.rFonts.set(qn('w:eastAsia'), '宋體')
        #table.cell(i+1, 0).text=singleTableData[i][1]
        #table.cell(i+1, 1).text=singleTableData[i][2]
        #table.cell(i+1, 2).text=singleTableData[i][3]
    

if __name__ == '__main__':
    #定義一個(gè)document
    document = Document()
    #設(shè)置字體默認(rèn)樣式
    document.styles['Normal'].font.name = u'宋體'
    document.styles['Normal']._element.rPr.rFonts.set(qn('w:eastAsia'), u'宋體')
    #獲取當(dāng)前庫(kù)下所有的表名信息和表注釋信息
    tableList = queryTableName()
    #循環(huán)查詢(xún)數(shù)據(jù)庫(kù),獲取表字段詳細(xì)信息,并調(diào)用generateWord,生成word數(shù)據(jù)
    #由于時(shí)間匆忙,我這邊選擇的是直接查詢(xún)數(shù)據(jù)庫(kù),執(zhí)行了100多次查詢(xún),可以進(jìn)行優(yōu)化,查詢(xún)出所有的表結(jié)構(gòu),在代碼里面將每個(gè)表結(jié)構(gòu)進(jìn)行拆分
    for singleTableName in tableList:
        data = query(singleTableName[0])
        generateWord(data,document,singleTableName[1])
    #保存至文檔
    document.save('數(shù)據(jù)庫(kù)設(shè)計(jì).docx')

3.生成的word文檔預(yù)覽

總結(jié)

運(yùn)行成功后會(huì)在目錄下的data文件夾中看到保存的Excel文件(以數(shù)據(jù)庫(kù)名為單位保存成文件),每個(gè)Excel第一個(gè)tab是空的(一個(gè)小bug暫未解決),其他每個(gè)tab以對(duì)應(yīng)表名進(jìn)行命名。

代碼很簡(jiǎn)單,供各位學(xué)習(xí)參考。

到此這篇關(guān)于利用Python批量導(dǎo)出mysql數(shù)據(jù)庫(kù)表結(jié)構(gòu)的文章就介紹到這了,更多相關(guān)Python批量導(dǎo)出mysql表結(jié)構(gòu)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

最新評(píng)論