Python3 で PostgreSQL の CRUD #Python3

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

接続情報

.env

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

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

version_check.py

#! /usr/bin/python
# -*- coding: utf-8 -*-
#
#	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
# -*- coding: utf-8 -*-
#
#	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
# -*- coding: utf-8 -*-
#
#	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
# -*- coding: utf-8 -*-
#
#	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")
#
# --------------------------------------------------------

Delete

postgre_delete.py

#! /usr/bin/python
# -*- coding: utf-8 -*-
#
#	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")
#
# --------------------------------------------------------

エラー対策

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
(省略）