BrandCultivation/dao/mysql_client.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine, text
from sqlalchemy.dialects.mysql import pymysql
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from config import database_config


class Mysql(object):
    def __init__(self):
        host = database_config.MYSQL_HOST
        port = database_config.MYSQL_PORT
        user = database_config.MYSQL_USER
        passwd = database_config.MYSQL_PASSWD
        dbname = database_config.MYSQL_DB
        self.engine = self._connect(host, port, user, passwd, dbname)
        self.engine = create_engine(self.engine)
        self._DBSession = sessionmaker(bind=self.engine)

    def _connect(self, host, port, user, pwd, db):
        client = "mysql+pymysql://" + user + ":" + pwd + "@" + host + ":" + str(port) + "/" + db
        return client

    @property
    def DBSession(self):
        return self._DBSession


if __name__ == '__main__':
    client = Mysql()
    sess = client.DBSession()
    sql_tables = text("show tables")
    for table in sess.execute(sql_tables).all():
        print(table[0])
        sql = text("""
            SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_COMMENT
            FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_NAME = :table_name;
        """)
        info = sess.execute(sql, {"table_name": table[0]})
        for tb in info:
            print(tb)