python连接Oracle数据库使用dsn

柔情只为你懂 2022-09-16 12:20 302阅读 0赞

from petal import Task
    from petal.util.db import DBEngineX
    import cx_Oracle
    from petal_tasks.huamu.sql_string import huamu_sql_zmy
    # from petal_tasks.collector_tools.collector_utils import get_template_client_base
    from petal_tasks.collector_tools.collector_utils_me import get_template_client_base
    
    
    class HMZMY(Task):
        def run(self):
            # 创建一个etl
            # 从数据库获取
            # 重命名
            # 存储到Collector
            (self.new_etl()
             .from_sql(sql=huamu_sql_zmy, dbe=self.get_db())
             # .rename({})
             .call_with_df(self.to_collector, "13f63902-******-40f3-b291-0059509bdd01")
             )
    
        @staticmethod
        def to_collector(df, template_id):
            breakpoint()
            data_mgr = get_template_client_base(template_uuid=template_id)
            data_mgr.data_delete(data_mgr.data_select())
            data_mgr.data_create(df, df_drop_columns=True)
    
        @staticmethod
        def get_db():
            host = '***********'
            port = ****
            user = '*******'
            passwd = '*****'
            service_name = 'orcl.wx'
            dsn = cx_Oracle.makedsn(host, port, service_name=service_name)
            dsn = f'oracle+cx_oracle://{user}:{passwd}@{dsn}'
            print('dsn: ', dsn)
            db = DBEngineX.from_url(dsn)
            res = db.execute("select count(*) from HUAMUJIEDAO.FLOWEVENT")
            print(res.fetchall())
            return db

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAMTAyNOWwj-elng_size_20_color_FFFFFF_t_70_g_se_x_16][]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAMTAyNOWwj-elng_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/85037d0647e04bac97997e55d65ab263.png