python中sqlalchemy连接已存在的数据库后进行查询 *** 作。

python中sqlalchemy连接已存在的数据库后进行查询 *** 作。

场景一，
只做已有数据库的映射关系。并且使用原生sql进行查询。

# -*- coding: utf-8 -*-
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.ext.declarative import Declarativemeta
from sqlalchemy import create_engine, metaData
from sqlalchemy.orm import sessionmaker

class DBCore(object):
    def __init__(self, map_tables: list = None):
        """
        映射db模型
        :param map_tables: list 默认无参数, 映射库里的所有表。如果tables有参数, 则只映射指定表，最好指定表，执行效率高
        """
        db_dict = {'user': conf['DB']['username'], 'password': conf['DB']['password'],
                   'host': conf['DB']['host'], 'port': conf['DB']['port'],
                   'database': conf['DB']['dbname'], 'charset': 'utf8mb4'}
        self.url = "mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset={charset}".format(**db_dict)
        self.engine = create_engine(self.url, pool_size=5, pool_recycle=1800, pool_pre_ping=True, encoding='utf-8')
        Session = sessionmaker(bind=self.engine)
        self.session = Session()

        self.tables = map_tables
        if not map_tables:
            self.auto_base = automap_base()
            self.auto_base.prepare(self.engine, reflect=True)
        else:
            self.metadata = metaData()
            self.metadata.reflect(self.engine, only=map_tables)
            self.auto_base = automap_base(metadata=self.metadata)
            self.auto_base.prepare()
if __name__ == '__main__':
    db = DBCore(['issues'])
    ##通过原生SQL语句 *** 作
    with db.session as cur:
        res = cur.execute('select * from issues where id = 1').fetchone()
        print(res)
        print(dict(res))
        print(res.id)

场景二
映射后，进行orm方式 *** 作数据库。

from sqlalchemy.ext.declarative import Declarativemeta
from sqlalchemy import create_engine, metaData
from sqlalchemy.orm import sessionmaker

def to_dict(self):
    return {c.name: getattr(self, c.name) for c in self.__table__.columns}


class DBCore(object):
    def __init__(self, map_tables: list = None):
        """
        映射db模型
        :param map_tables: list 默认无参数, 映射库里的所有表。如果tables有参数, 则只映射指定表
        """
        db_dict = {'user': conf['DB']['username'], 'password': conf['DB']['password'],
                   'host': conf['DB']['host'], 'port': conf['DB']['port'],
                   'database': conf['DB']['dbname'], 'charset': 'utf8mb4'}
        self.url = "mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset={charset}".format(**db_dict)
        self.engine = create_engine(self.url, pool_size=5, pool_recycle=1800, pool_pre_ping=True, encoding='utf-8')
        Session = sessionmaker(bind=self.engine)
        self.session = Session()

        self.tables = map_tables
        if not map_tables:
            self.auto_base = automap_base()
            self.auto_base.prepare(self.engine, reflect=True)
        else:
            self.metadata = metaData()
            self.metadata.reflect(self.engine, only=map_tables)
            self.auto_base = automap_base(metadata=self.metadata)
            self.auto_base.prepare()

    def get_model(self, table_name: str):
        """
        根据数据库已存在的表名动态映射模型类
        """
        if not table_name:
            return None
        try:
            model: Declarativemeta = getattr(self.auto_base.classes, table_name)
            model.to_dict = to_dict
            return model
        except AttributeError:
            raise
            
if __name__ == '__main__':
    db = DBCore()
    #orm *** 作
    issues= db.get_model("issues")
    res = db.session.query(issues).filter(issues.id == 1).one()
    print(res.title)
    res_dict = res.to_dict()
    print(res_dict)
    db.session.close()

think：
场景一， *** 作简单，直接连接想要的表就行，sql写起来也简单。
场景二，映射全部的表后，根据get_model方法获取指定的表，然后进行查询等 *** 作。这样执行速度较慢，当然在初始化时可以指定几张表，然后 *** 作，也快。