Python3 で PostgreSQL の CRUD

Last updated at 2024-06-08Posted at 2019-05-19

こちらのページと同じことを Python3 で行いました。
Ruby で PostgreSQL の CRUD

接続情報

.env

user='****'
password='*****'
data_base='city'

PostgreSQL　のバージョンを調べる方法です。

version_check.py

#! /usr/bin/python
#
#	version_check.py
#					Nov/17/2019
#
# --------------------------------------------------------
import os
import sys
import psycopg2
import psycopg2.extras
from dotenv import load_dotenv
#
sys.stderr.write ("*** 開始 ***\n")
dotenv_path = '.env'
load_dotenv(dotenv_path)
USER=os.environ.get("user")
PASSWORD=os.environ.get("password")
DATA_BASE=os.environ.get("data_base")
#
conn = psycopg2.connect("dbname=" + DATA_BASE + " user=" + USER + " password=" + PASSWORD)
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
#
sql_str = "SELECT VERSION()"
cur.execute(sql_str)
#rows = cur.fetchall()
rows = cur.fetchone()
print(rows[0])
#
cur.close()
conn.close()
#
sys.stderr.write ("*** 終了 ***\n")
#
# --------------------------------------------------------

実行結果

$ ./version_check.py 
*** 開始 ***
PostgreSQL 14.8 (Ubuntu 14.8-0ubuntu0.22.10.1) on x86_64-pc-linux-gnu, compiled by gcc (Ubuntu 12.2.0-3ubuntu1) 12.2.0, 64-bit
*** 終了 ***

Create

postgre_create.py

#! /usr/bin/python3
#
#	postgre_create.py
#
#					May/19/2019
#
# --------------------------------------------------------
import	os
import	sys
import	psycopg2
from dotenv import load_dotenv
#
# --------------------------------------------------------
def dict_append_proc(dict_aa,key,name,population,date_mod):
	dict_aa[key] = {'name':name,'population':population,'date_mod':date_mod}
#
	return dict_aa
#
# --------------------------------------------------------
def	data_prepare_proc ():
#
	dict_aa = {} 
#
	dict_aa = dict_append_proc (dict_aa,'t3461','広島',25176,'2003-9-24')
	dict_aa = dict_append_proc (dict_aa,'t3462','福山',47935,'2003-5-15')
	dict_aa = dict_append_proc (dict_aa,'t3463','東広島',28654,'2003-2-8')
	dict_aa = dict_append_proc (dict_aa,'t3464','呉',83152,'2003-10-9')
	dict_aa = dict_append_proc (dict_aa,'t3465','尾道',42791,'2003-8-4')
	dict_aa = dict_append_proc (dict_aa,'t3466','竹原',35687,'2003-1-21')
	dict_aa = dict_append_proc (dict_aa,'t3467','三次',81296,'2003-7-23')
	dict_aa = dict_append_proc (dict_aa,'t3468','大竹',21764,'2003-10-26')
	dict_aa = dict_append_proc (dict_aa,'t3469','府中',75428,'2003-12-15')
#
	return	dict_aa
#
# --------------------------------------------------------
def	create_table_proc (cursor_aa):
	sql_str="create table cities (id varchar(10) primary key, name varchar(20)," \
		+ " population int, date_mod date)"
#
	try:
		cursor_aa.execute (sql_str)
	except Exception as ee:
		sys.stderr.write ("*** error *** create_table_proc ***\n")
		sys.stderr.write (str (ee) + "\n")
#
# --------------------------------------------------------
def	drop_table_proc (cursor_aa):
	sql_str=u"drop table cities"
	try:
		cursor_aa.execute (sql_str)
	except Exception as ee:
		sys.stderr.write ("*** error *** drop_table_proc ***\n")
		sys.stderr.write (str (ee) + "\n")
#
# --------------------------------------------------------
def	sql_insert_proc	(cursor_aa,id_in,name_in,ipop_in,date_mod_in):
#
	ft_aa="insert into cities (id,name,population,date_mod) values ("
	ft_bb ="'%s','%s',%d,'%s')" % (id_in,name_in,ipop_in,date_mod_in)
	sql_str=ft_aa + ft_bb
	cursor_aa.execute (sql_str)
#
# --------------------------------------------------------
def	table_insert_proc	(cursor,dict_aa):
	for key in dict_aa:
		unit = dict_aa[key]
		try:
			sql_insert_proc	(cursor,key,unit['name'], \
				unit['population'],unit['date_mod'])
		except Exception as ee:
			sys.stderr.write ("*** error *** table_insert_proc ***\n")
			sys.stderr.write (str (ee) + "\n")
			sys.stderr.write (key + "\n")
# --------------------------------------------------------
sys.stderr.write ("*** 開始 ***\n")
#
#
dict_aa = data_prepare_proc ()
#
dotenv_path = '.env'
load_dotenv(dotenv_path)
USER=os.environ.get("user")
PASSWORD=os.environ.get("password")
DATA_BASE=os.environ.get("data_base")
#
conn = psycopg2.connect("dbname=" + DATA_BASE + " user=" + USER + " password=" + PASSWORD)
cur = conn.cursor()
drop_table_proc (cur)
create_table_proc (cur)
table_insert_proc (cur,dict_aa)
#
conn.commit ()
#
#
cur.close ()
conn.close ()
#
sys.stderr.write ("*** 終了 ***\n")
#
# --------------------------------------------------------

Read

postgre_read.py

#! /usr/bin/python
#
#	postgre_read.py
#					May/19/2019
#
# --------------------------------------------------------
import os
import sys
import psycopg2
import psycopg2.extras
from dotenv import load_dotenv
#
sys.stderr.write ("*** 開始 ***\n")
dotenv_path = '.env'
load_dotenv(dotenv_path)
USER=os.environ.get("user")
PASSWORD=os.environ.get("password")
DATA_BASE=os.environ.get("data_base")
#
conn = psycopg2.connect("dbname=" + DATA_BASE + " user=" + USER + " password=" + PASSWORD)
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
#
sql_str = "select * from cities order by ID"
cur.execute(sql_str)
rows = cur.fetchall()
for row in rows:
	print(row['id'],row['name'],row['population'],row['date_mod'])
#
cur.close()
conn.close()
#
sys.stderr.write ("*** 終了 ***\n")
#
# --------------------------------------------------------

Update

postgre_update.py

#! /usr/bin/python3
#
#	postgre_update.py
#
#				May/19/2019
#
# --------------------------------------------------------
import os
import sys
import string
import	datetime
from time import localtime,strftime
import psycopg2
from dotenv import load_dotenv
#
#
# --------------------------------------------------------
def	sql_update_string_gen_proc	(id_in,ipop_in):
	date_mod = strftime ("%Y-%m-%d",localtime ())
	ft_aa=u"update cities set population = %d, " % ipop_in
	ft_bb=u"date_mod='%s' " % date_mod
	ft_cc=u"where id = '%s'" % id_in
	sql_str=ft_aa + ft_bb + ft_cc
	print (sql_str)
#
	return	sql_str
# --------------------------------------------------------
sys.stderr.write ("*** 開始 ***\n")
key_in = sys.argv[1]
population_in = int (sys.argv[2])
print ("%s\t%d" % (key_in, population_in))
#
dotenv_path = '.env'
load_dotenv(dotenv_path)
USER=os.environ.get("user")
PASSWORD=os.environ.get("password")
DATA_BASE=os.environ.get("data_base")
#
conn = psycopg2.connect("dbname=" + DATA_BASE + " user=" + USER + " password=" + PASSWORD)
#
cur = conn.cursor()
#
sql_str=sql_update_string_gen_proc (key_in,population_in)
cur.execute(sql_str)
conn.commit ()
#
cur.close ()
conn.close ()
#
sys.stderr.write ("*** 終了 ***\n")
#
# --------------------------------------------------------

実行方法

./postgre_update.py t3462 4561200

Delete

postgre_delete.py

#! /usr/bin/python
#
#	postgre_delete.py
#
#				May/19/2019
#
# --------------------------------------------------------
import os
import sys
import psycopg2
from dotenv import load_dotenv
#
# --------------------------------------------------------
sys.stderr.write ("*** 開始 ***\n")
key_in = sys.argv[1]
print ("%s" % key_in)
#
dotenv_path = '.env'
load_dotenv(dotenv_path)
USER=os.environ.get("user")
PASSWORD=os.environ.get("password")
DATA_BASE=os.environ.get("data_base")
#
conn = psycopg2.connect("dbname=" + DATA_BASE + " user=" + USER + " password=" + PASSWORD)
cur = conn.cursor()
#
sql_str=u"delete from cities where id = '%s'" % key_in
cur.execute (sql_str)
#
conn.commit ()
#
cur.close ()
conn.close ()
#
#
sys.stderr.write ("*** 終了 ***\n")
#
# --------------------------------------------------------

実行方法

./postgre_delete.py t3465

エラー対策

psycopg2.OperationalError: connection to server on socket "/var/run/postgresql/.s.PGSQL.5432" failed: FATAL:  Peer authentication failed for user "scott"

peer を trust に変えます。

/etc/postgresql/14/main/pg_hba.conf

(省略）
#local   all             all                                     peer
local   all             all                                     trust
(省略）

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up