50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
|
import os
|
||
|
|
|
||
|
|
import psycopg2
|
||
|
|
from psycopg2.extras import RealDictCursor
|
||
|
|
|
||
|
|
_pool = None
|
||
|
|
|
||
|
|
|
||
|
|
def init_pool() -> None:
|
||
|
|
global _pool
|
||
|
|
if _pool is None:
|
||
|
|
_pool = {
|
||
|
|
"host": os.getenv("DB_HOST", "10.4.7.18"),
|
||
|
|
"user": os.getenv("DB_USER", "aissa"),
|
||
|
|
"password": os.getenv("DB_PASSWORD", "Foodlinkk#2026"),
|
||
|
|
"dbname": os.getenv("DB_NAME", "foodlinkk"),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _conn():
|
||
|
|
init_pool()
|
||
|
|
return psycopg2.connect(**_pool, cursor_factory=RealDictCursor)
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_one(sql: str, params: tuple = ()) -> dict | None:
|
||
|
|
with _conn() as conn:
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute(sql, params)
|
||
|
|
row = cur.fetchone()
|
||
|
|
conn.commit()
|
||
|
|
return dict(row) if row else None
|
||
|
|
|
||
|
|
|
||
|
|
def fetch_all(sql: str, params: tuple = ()) -> list:
|
||
|
|
with _conn() as conn:
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute(sql, params)
|
||
|
|
rows = cur.fetchall()
|
||
|
|
conn.commit()
|
||
|
|
return [dict(r) for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
def execute_returning(sql: str, params: tuple = ()) -> dict:
|
||
|
|
with _conn() as conn:
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute(sql, params)
|
||
|
|
row = cur.fetchone()
|
||
|
|
conn.commit()
|
||
|
|
return dict(row) if row else {}
|