python3 通过sqlalchemy使用原生sql语句查询数据库

浅浅的花香味﹌ 2022-10-10 01:45 291阅读 0赞

## class demo ##

class DbBase:
        def __init__(self, db_choice: str, db_user: str, db_password: str, host: str, port: int, db_name: str,
                     pool_size=int(5), max_overflow=int(20),
                     pool_recycle=int(120):
            self.db_choice = db_choice  # 使用的数据库，可以传值为oracle，mysql，pg
            self.user = db_user  # 数据库用户名
            self.password = db_password  # 数据库密码
            self.host = host  # 数据库ip
            self.port = port  # 数据库端口
            self.db_name = db_name  # 数据库名称（oracle下为sid）
            self.echo = bool(SQLALCHEMY_ECHO)  # 是不是要把所执行的SQL打印出来，一般用于调试
            self.pool_size = pool_size  # 连接池大小
            self.max_overflow = max_overflow  # 连接池最大的大小
            self.pool_recycle = pool_recycle  # 多久时间主动回收连接
            self.oracle_uri = f"oracle+cx_oracle://{self.user}:{self.password}@{self.host}:{self.port}/{self.db_name}"
            self.mysql_uri = f"mysql+pymysql://{self.user}:{self.password}@{self.host}:{self.port}/{self.db_name}"
            self.pg_uri = f"postgresql+psycopg2://{self.user}:{self.password}@{self.host}:{self.port}/{self.db_name}"
            self.choice_uri_dict = { "oracle": self.oracle_uri,
                                    "mysql": self.mysql_uri,
                                    "pg": self.pg_uri}
            self.session = self.get_session()  # 链接会话
    
        def fetch_db_data(self, sql):
            """ 通过sql语句查询数据库中的数据 :param sql: sql语句 :return: """
            try:
                with self.user_session() as s:
                    db_data = s.execute(text(sql)).fetchall()
                    return True, db_data
            except Exception as e:
                return False, e
    
        def get_session(self):
            """ 获取上下文对象session :return:session """
            engine = create_engine(
                self.choice_uri_dict.get(self.db_choice),
                echo=self.echo,
                pool_size=self.pool_size,
                max_overflow=self.max_overflow,
                pool_recycle=self.pool_recycle,
            )
            session = sessionmaker(bind=engine)
            return session
    
        @contextlib.contextmanager
        def user_session(self):
            """ 使用session对数据库进行操作，支持上下文 :return: session上下文对象 """
            self.s = self.session()
            try:
                yield self.s
                self.s.commit()
            except Exception as e:
                self.s.rollback()
                raise e
            finally:
                self.s.close()