利用sqlalchemy封装一个查询类

通过此查询类可实现所有数据查询和单条数据查询，并将数据封装成{‘字段名’:’字段值’}的形式返回。

依赖库版本：

cx-Oracle==8.3.0
Flask-SQLAlchemy==2.5.1
SQLAlchemy==2.0.30

数据库连接方法：

def oracle_connect(username, password, db):
    """oracle 连接"""
    try:
        engine = create_engine(
            'oracle+cx_oracle://{}:{}@{}'.format(username, password, db),
            pool_size=20,
            pool_recycle=-1,
            pool_pre_ping=True,
            pool_timeout=15,

        )
        engine.max_identifier_length = 30
        session_factory = sessionmaker(bind=engine)
        session = session_factory()
    except Exception as e:
        """
        在这里做一些异常处理
        """
        return None

    return session

查询类

class GetDBData(object):
    def __init__(self, db_conn):
        self.db_conn = db_conn

    def fetchall(self, sql):
        """
        查询所有数据，并转换成字典
        :param sql: 需要执行的sql
        :return:
        """
        query = self.db_conn.execute(text(sql)).all()
        # 将数据转换成字典返回
        dict_results = [row._asdict() for row in query]
        return dict_results

    def fetchone(self, sql):
        """
        查询单条数据，并转换成字典
        :param sql: 需要执行的sql
        :return:
        """
        query = self.db_conn.execute(text(sql)).first()
        if query:
            # 将数据转换成字典返回
            dict_results = query._asdict()
        else:
            dict_results = {}
        return dict_results

    def __del__(self):
        """
        每次执行完成后主动关闭数据库连接
        :return:
        """
        self.db_conn.close()
        print("关闭数据库连接")

每次执行完成后自动关闭数据库连接（这里关闭的是session连接）

使用时注意依赖库版本，低版本依赖库可能不支持