python">import pandas as pd
import numpy as np
import re
import psycopg2
from sqlalchemy import create_engine

1. 连接数据库，下载所需数据

python">def download_sqlfile(sql_file):
    """
    连接数据库，下载所需数据
    """
    # 建立数据库连接
    con = psycopg2.connect(database="warehouse", user="XXXX", password="XXXXX", host="XX.XX.cn", port="5432")
    # 调用游标对象
    cur = con.cursor()

    # 执行语句，下载数据
    sql_command = "select * from %s"%sql_file # 
    # 获取数据
    # 由于数据库首次运行容易丢失连接，需两次重复运行
    try:
        data = pd.read_sql(sql_command, con,params=(sql_file,))
        print('连接数据库成功')
    except:
        print('首次连接数据库失败，尝试第二次连接数据库')
        data = pd.read_sql(sql_command, con,params=(sql_file,))
        print('连接数据库成功')
    
    # 关闭数据库连接关闭游标
    cur.close()
    con.close()
    
    return data

读取示例

python"># 读取10月数据
sql_file_10 = 'staging_sy_data.bt_51job_2023102008_20231016143722'
df_10 = download_sqlfile(sql_file_10)

2. 数据上传到数据库

python"># 数据上传到数据库
def upload_table(table_name, table):
    """
    将本地Dataframe存储到数据库中
    """
    # 数据库连接信息 database="warehouse", user="XX", password="XXX", host="XX.XX.cn", port="5432")
    db_connection = {
        'host': 'XX.XX.cn',
        'port': '5432',
        'dbname': 'warehouse',
        'user': 'XXX',
        'password': 'XXXXX'
    }

    # 创建数据库引擎
    engine = create_engine('postgresql+psycopg2://{user}:{password}@{host}:{port}/{dbname}'.format(**db_connection))
    
    # 将本地Dataframe存储到数据库中
    with engine.connect() as conn:
        print(bool(conn))
        
        table.to_sql(name = table_name, con = engine, schema = 'staging_data_sy') # 注意此处to_sql需要使用sqlalchemy 包的 create_engine；数据库链接只能用engine不能使用
        
#     table.to_sql(table_name, engine, if_exists='replace', index=False)
    
    print('表格上传成功')
    
    return None

数据上传示例：

python"># 将data存储到数据库中
table_name = 'bt_51job_chengdu_202310'
try:
    upload_table(table_name, df_chengdu)
except:
    print('第一次上传失败，正在进行第二次上传...')
    upload_table(table_name, df_chengdu)