39 lines
869 B
Python
39 lines
869 B
Python
|
"""PostgreSQL connection helpers."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
|
||
|
|
import psycopg2
|
||
|
|
from psycopg2.extensions import connection
|
||
|
|
|
||
|
|
|
||
|
|
def db_params() -> dict[str, str]:
|
||
|
|
return {
|
||
|
|
"host": os.getenv("DB_HOST", "localhost"),
|
||
|
|
"port": os.getenv("DB_PORT", "5432"),
|
||
|
|
"dbname": os.getenv("DB_NAME", "kairo_dev"),
|
||
|
|
"user": os.getenv("DB_USER", "kairo_dev"),
|
||
|
|
"password": os.getenv("DB_PASSWORD", "dev_password"),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def get_connection() -> connection:
|
||
|
|
p = db_params()
|
||
|
|
return psycopg2.connect(
|
||
|
|
host=p["host"],
|
||
|
|
port=p["port"],
|
||
|
|
database=p["dbname"],
|
||
|
|
user=p["user"],
|
||
|
|
password=p["password"],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def check_db() -> bool:
|
||
|
|
conn = get_connection()
|
||
|
|
try:
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute("SELECT 1")
|
||
|
|
cur.fetchone()
|
||
|
|
return True
|
||
|
|
finally:
|
||
|
|
conn.close()
|