Python+PostgreSQL：创建数据库并从Pandas DataFrame填充表求助

别担心基础问题呀，新手入门都是从这些步骤开始的～下面我会一步步带你实现你的需求，代码里也会加注释方便你理解。

1. 准备工作：安装必要的Python库 #

首先得安装几个关键的工具库，打开终端运行以下命令：

pip install psycopg2-binary pandas configparser

• psycopg2-binary：PostgreSQL的Python官方驱动，负责和数据库建立交互
• pandas：处理CSV文件和DataFrame的核心工具
• configparser：用来读取你的.ini配置文件

2. 先确认你的.ini文件格式（示例参考） #

假设你的.ini文件（比如命名为db_config.ini）内容是这样的，如果你的格式不同，后续代码里对应调整字段就行：

[postgresql]
host=localhost
port=5432
user=your_username
password=your_password

3. 读取配置并创建testdb数据库 #

这里有个小细节：创建新数据库时，我们得先连接到PostgreSQL默认的postgres数据库（因为testdb还不存在，没法直接连接它）。代码如下：

import configparser
import psycopg2
from psycopg2 import OperationalError

def create_database(db_name):
    # 读取.ini配置文件
    config = configparser.ConfigParser()
    config.read('db_config.ini')
    
    try:
        # 先连接到默认的postgres数据库
        conn = psycopg2.connect(
            host=config['postgresql']['host'],
            port=config['postgresql']['port'],
            user=config['postgresql']['user'],
            password=config['postgresql']['password'],
            dbname='postgres'  # 指定默认数据库
        )
        conn.autocommit = True  # 必须开启自动提交才能执行创建数据库的语句
        cur = conn.cursor()
        
        # 先检查数据库是否已存在，避免重复创建报错
        cur.execute(f"SELECT 1 FROM pg_database WHERE datname = '{db_name}'")
        exists = cur.fetchone()
        if not exists:
            cur.execute(f"CREATE DATABASE {db_name}")
            print(f"数据库 {db_name} 创建成功！")
        else:
            print(f"数据库 {db_name} 已经存在啦～")
        
        # 关闭游标和连接
        cur.close()
        conn.close()
    except OperationalError as e:
        print(f"数据库连接或创建出错：{e}")

# 调用函数创建testdb
create_database('testdb')

4. 读取CSV文件为Pandas DataFrame #

这一步非常简单，用pandas的read_csv方法就能搞定，记得替换成你的CSV文件路径：

import pandas as pd

def read_csv_to_df(csv_path):
    try:
        df = pd.read_csv(csv_path)
        print(f"CSV文件读取成功，共 {len(df)} 行数据")
        return df
    except FileNotFoundError:
        print(f"找不到指定的CSV文件：{csv_path}")
        return None

# 替换成你的CSV文件实际路径，比如 './data.csv'
df = read_csv_to_df('your_csv_file.csv')

5. 在testdb中创建表并填充DataFrame数据 #

这里我们用SQLAlchemy引擎来配合pandas的to_sql方法，会比直接用psycopg2更方便，能自动创建表并填充数据：

from sqlalchemy import create_engine

def df_to_postgresql(df, db_name, table_name):
    if df is None:
        print("没有可用的DataFrame数据，无法写入数据库")
        return
    
    # 再次读取配置
    config = configparser.ConfigParser()
    config.read('db_config.ini')
    
    # 创建SQLAlchemy连接引擎
    engine_url = f"postgresql+psycopg2://{config['postgresql']['user']}:{config['postgresql']['password']}@{config['postgresql']['host']}:{config['postgresql']['port']}/{db_name}"
    engine = create_engine(engine_url)
    
    try:
        # 写入数据，if_exists参数可按需调整：
        # 'fail'（默认，表存在则报错）、'replace'（替换现有表）、'append'（追加数据）
        df.to_sql(
            name=table_name,
            con=engine,
            if_exists='replace',
            index=False  # 不要把DataFrame的索引作为数据库表的列
        )
        print(f"数据成功写入 {db_name} 数据库的 {table_name} 表中！")
    except Exception as e:
        print(f"写入数据库出错：{e}")

# 调用函数，替换成你想要的表名，比如 'test_table'
df_to_postgresql(df, 'testdb', 'your_table_name')

可选：整合所有步骤成一个脚本 #

你可以把上面的函数整合到一个脚本里，按顺序执行所有操作：

import configparser
import psycopg2
from psycopg2 import OperationalError
import pandas as pd
from sqlalchemy import create_engine

def create_database(db_name):
    # 同上面的函数代码
    config = configparser.ConfigParser()
    config.read('db_config.ini')
    try:
        conn = psycopg2.connect(
            host=config['postgresql']['host'],
            port=config['postgresql']['port'],
            user=config['postgresql']['user'],
            password=config['postgresql']['password'],
            dbname='postgres'
        )
        conn.autocommit = True
        cur = conn.cursor()
        cur.execute(f"SELECT 1 FROM pg_database WHERE datname = '{db_name}'")
        exists = cur.fetchone()
        if not exists:
            cur.execute(f"CREATE DATABASE {db_name}")
            print(f"数据库 {db_name} 创建成功！")
        else:
            print(f"数据库 {db_name} 已经存在啦～")
        cur.close()
        conn.close()
    except OperationalError as e:
        print(f"数据库连接或创建出错：{e}")

def read_csv_to_df(csv_path):
    # 同上面的函数代码
    try:
        df = pd.read_csv(csv_path)
        print(f"CSV文件读取成功，共 {len(df)} 行数据")
        return df
    except FileNotFoundError:
        print(f"找不到指定的CSV文件：{csv_path}")
        return None

def df_to_postgresql(df, db_name, table_name):
    # 同上面的函数代码
    if df is None:
        print("没有可用的DataFrame数据，无法写入数据库")
        return
    config = configparser.ConfigParser()
    config.read('db_config.ini')
    engine_url = f"postgresql+psycopg2://{config['postgresql']['user']}:{config['postgresql']['password']}@{config['postgresql']['host']}:{config['postgresql']['port']}/{db_name}"
    engine = create_engine(engine_url)
    try:
        df.to_sql(
            name=table_name,
            con=engine,
            if_exists='replace',
            index=False
        )
        print(f"数据成功写入 {db_name} 数据库的 {table_name} 表中！")
    except Exception as e:
        print(f"写入数据库出错：{e}")

if __name__ == "__main__":
    # 1. 创建testdb
    create_database('testdb')
    # 2. 读取CSV
    df = read_csv_to_df('your_csv_file.csv')
    # 3. 写入数据库表
    if df is not None:
        df_to_postgresql(df, 'testdb', 'test_table')

内容的提问来源于stack exchange，提问作者qstnr