python postgresql从pandas数据框创建数据库并填充表

Posted 2023-02-16

【中文标题】python postgresql从pandas数据框创建数据库并填充表

【英文标题】：python postgresql create database and populate table from pandas dataframe

【发布时间】：2018-08-13 16:58:54

【问题描述】：

我对 Python 很陌生，对使用 postgresql 也很陌生，所以如果这是基本的东西（到目前为止，我未能实现），请原谅我。我正在尝试编写一个python代码：

创建一个新数据库 (testdb) 将 csv 文件读入 pandas 数据帧 根据 pandas 数据框在数据库中创建并填充一个新表。

到目前为止，我有 3 个不同的文件：a) 一个 .ini-文件，用于存储创建新数据库所需的数据库信息，b) 一个 .csv-文件（来自 here，名为 @987654328 @) 和 c) 我的 python 代码。

database.ini：

[postgresql]
host=localhost
user=postgres
password=creator
port=5432

db_creator.py：

from config import config
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database
import pandas as pd

# useful info for psycopg2:
# https://***.com/questions/34484066/create-a-postgres-database-using-python


class MyDB(object):
    def __init__(self):
        self.params = config()

    def create_new_db(self, newdb):
        user, host, port = self.params['user'], self.params['host'], testdb.params['port']
        pw = self.params['password']
        url = 'postgresql://:@:/'
        url = url.format(user, pw, host, port, newdb)

        engine = create_engine(url)
        if not database_exists(engine.url):
            create_database(engine.url)

        print(database_exists(engine.url))


if __name__ == '__main__':

    testdb = MyDB()
    testdb.create_new_db('testdb')

当我尝试这样做时，我收到以下错误：

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError)

但是，当我按照this SO post 中的建议进行操作时，它会起作用。不幸的是，这篇文章中的答案使用psycopg2 创建一个新数据库，但我想用sqlalchemy 来做这件事（也是因为我认为使用sqlalchemy 进一步使用Pandas 数据框会更容易（例如显示here。还是我错了？）。我认为当使用sqlqlchemy 执行此操作时，应该可以通过以下方式将数据从 csv 文件读取到 pandas 数据帧中，然后在新数据库：

def connect_alchemy(user, host, port, db, password):
    url = 'postgresql://:@:/'
    url = url.format(user, password, host, port, db)
    con = sqlalchemy.create_engine(url, client_encoding='utf8')
    mydata = pd.read_csv('100_recs.csv', delimiter=';', quotechar='"')
    data_db = mydata.to_sql(name='100_records', con=con, if_exists='replace', index=True, chunksize=10)

    print(con.execute('SELECT * from 100_records'))

但老实说，我被困在这里需要一些帮助......如果有人能指出我正确的方向，那就太好了。

编辑： 啊愚蠢的我！ 所以我在db_creator.py 的以下几行中有一个旧错字

user, host, port = testdb.params['user'], testdb.params['host'], testdb.params['port']
        pw = testdb.params['password']

应该是：

user, host, port = self.params['user'], self.params['host'], self.params['port']
        pw = self.params['password']

我已经改变了这个。

然后我也忘记在此处添加config.py 文件。对此深表歉意。

给你：

config.py

# source: http://www.postgresqltutorial.com/postgresql-python/connect/
from configparser import ConfigParser


def config(filename='database.ini', section='postgresql'):
    # create a parser
    parser = ConfigParser()
    # read config file
    parser.read(filename)

    # get section, default to postgresql
    db = 
    if parser.has_section(section):
        params = parser.items(section)
        for param in params:
            db[param[0]] = param[1]
    else:
        raise Exception('Section 0 not found in the 1 file'.format(section, filename))

    return db

编辑 2：

现在可以使用以下设置：

database.ini：

[postgresql]
host=localhost
user=postgres
password=postgres
port=5432

config.py：

# source: http://www.postgresqltutorial.com/postgresql-python/connect/
from configparser import ConfigParser


def config(filename='database.ini', section='postgresql'):
    # create a parser
    parser = ConfigParser()
    # read config file
    parser.read(filename)

    # get section, default to postgresql
    db = 
    if parser.has_section(section):
        params = parser.items(section)
        for param in params:
            db[param[0]] = param[1]
    else:
        raise Exception('Section 0 not found in the 1 file'.format(section, filename))

    return db

csv 文件：来自here

db_creator.py

from config import config
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database
import pandas as pd

# useful info for psycopg2:
# https://***.com/questions/34484066/create-a-postgres-database-using-python


class MyDB(object):
    def __init__(self):
        self.params = config()

    def create_new_db(self, newdb):
        user, host, port = self.params['user'], self.params['host'], self.params['port']
        pw = self.params['password']
        url = 'postgresql://:@:/'
        url = url.format(user, pw, host, port, newdb)

        self.engine = create_engine(url, client_encoding='utf8')
        if not database_exists(self.engine.url):
            create_database(self.engine.url)
        # print(database_exists(engine.url))

def df2postgres(engine, df):
    con = engine.connect()
    df.to_sql(name='records', con=con, if_exists='replace', index=True, chunksize=10)

    return con



if __name__ == '__main__':

    testdb = MyDB()
    testdb.create_new_db('testdb')
    engn = testdb.engine
    df = pd.read_csv('100_recs.csv', delimiter=';', quotechar='"', encoding='utf-8')
    con = df2postgres(engine=engn, df=df)
    dta = con.execute('SELECT * FROM records LIMIT 5;')
    print(dta.fetchall())

为愚蠢的错误道歉......

【参考方案1】：

psycopg2 只是一个模块，它提供了一个适配器来使用python 代码连接到Postgres dbs。另一方面，SQLAlcehmy 是Object Relational Mapper。它提供了一个额外的抽象层，将 sql 表镜像到 python 对象并标准化操作以在代码和数据库之间移动数据。

您可以同时使用psycopg2 和sqlalchemy，如您在documentation 中看到的那样

from sqlalchemy import create_engine

engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')