PostgreSQL前提であれば psycopg2での接続が一般的
import psycopg2
from datetime import datetime
from decimal import Decimal
def fetchone_dict(table_name):
# PostgreSQLへの接続
conn = psycopg2.connect(
dbname="soodb",
user="sooni",
password="soopass",
host="vm102",
port="5432"
)
try:
# カーソルを作成
with conn.cursor() as cur:
# SELECT文の発行 (テーブル名をパラメータとして使用)
query = f"SELECT * FROM {table_name}"
cur.execute(query)
# 1件のみ取得
result = cur.fetchone()
if result:
# カラム名を取得
column_names = [desc[0] for desc in cur.description]
# 結果を辞書形式に変換
result_dict = dict(zip(column_names, result))
# タイムスタンプ型およびDecimal型のカラムを文字列に変換
for col_name, value in result_dict.items():
if isinstance(value, datetime):
result_dict[col_name] = value.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(value, Decimal):
result_dict[col_name] = str(value)
# 辞書型の結果を返す
return result_dict
else:
return None
finally:
# 接続を閉じる
conn.close()
# 使用例: テーブル名を指定し、結果を取得して表示
result = fetchone_dict("fruits")
if result:
print(result)
else:
print("No result found")
PostgreSQL前提であれば psycopg2パッケージ特有の記載もあります
import psycopg2
import psycopg2.extras
from datetime import datetime
from decimal import Decimal
def fetchone_dict(table_name):
# PostgreSQLへの接続
conn = psycopg2.connect(
dbname="soodb",
user="sooni",
password="soopass",
host="vm102",
port="5432"
)
try:
# 結果を辞書形式で取得するカーソルを作成
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
# SELECT文の発行 (テーブル名をパラメータとして使用)
query = f"SELECT * FROM {table_name}"
cur.execute(query)
# 1件のみ取得
result_dict = cur.fetchone()
if result_dict:
# タイムスタンプ型およびDecimal型のカラムを文字列に変換
for col_name, value in result_dict.items():
if isinstance(value, datetime):
result_dict[col_name] = value.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(value, Decimal):
result_dict[col_name] = str(value)
# 辞書型の結果を返す
return dict(result_dict)
else:
return None
finally:
# 接続を閉じる
conn.close()
# 使用例: テーブル名を指定し、結果を取得して表示
result = fetchone_dict("fruits")
if result:
print(result)
else:
print("No result found")
jaydebeapiによるJDBC接続の場合
import jaydebeapi
from datetime import datetime
# JDBC接続のための設定
jdbc_url = "jdbc:postgresql://vm102:5432/soodb"
jdbc_driver_name = "org.postgresql.Driver"
jdbc_driver_loc = r"C:\python-prj\lib\postgresql-42.7.4.jar"
connection_args = {
'user': 'sooni',
'password': 'soopass'
}
# JDBC接続
conn = jaydebeapi.connect(jdbc_driver_name, jdbc_url, connection_args, jdbc_driver_loc)
# カーソルを作成し、クエリを実行
with conn.cursor() as cur:
# SELECT文の発行
cur.execute("SELECT * FROM fruits WHERE lno = ?", (1,))
# 1件のみ取得
result = cur.fetchone()
# 結果が存在する場合、Key:value形式に変換
if result:
# カラム名を取得
column_names = [desc[0] for desc in cur.description]
# 結果を辞書形式に変換
result_dict = dict(zip(column_names, result))
# タイムスタンプ型のカラムを文字列に変換
for col_name, value in result_dict.items():
if isinstance(value, datetime):
result_dict[col_name] = value.strftime('%Y-%m-%d %H:%M:%S')
print(result_dict)
else:
print("No result found")
# 接続を閉じる
conn.close()
検証用データ
CREATE TABLE fruits (
fruit_id SERIAL PRIMARY KEY, -- 自動的にインクリメントされるフルーツID
fruit_name VARCHAR(100), -- フルーツ名
stock_quantity INT, -- 在庫数
price NUMERIC(6, 2), -- 価格(小数点以下2桁)
is_available BOOLEAN, -- 販売中かどうか
last_updated TIMESTAMP -- 最終更新日時
);
truncate table fruits;
INSERT INTO fruits (fruit_name, stock_quantity, price, is_available, last_updated) VALUES
('Apple', null, 1.20, TRUE, '2024-08-01 10:00:00'),
('Banana', 200, 0.50, TRUE, '2024-08-02 11:30:00'),
('Cherry', 50, 3.00, FALSE, '2024-08-03 14:00:00'),
('Orange', 120, 1.10, TRUE, '2024-08-04 09:45:00'),
('Strawberry', null, 2.50, TRUE, '2024-08-05 08:15:00');