40 lines
1 KiB
Python
40 lines
1 KiB
Python
from os import environ
|
|
from collections import namedtuple
|
|
import pg8000
|
|
|
|
from flask import g
|
|
|
|
DBResult = namedtuple('Result', 'rows rowcount')
|
|
|
|
def db():
|
|
if not hasattr(g, 'db_connection'):
|
|
g.db_connection = pg8000.connect(
|
|
database=environ['DB_NAME'],
|
|
user=environ['DB_USER'],
|
|
password=environ['DB_PASSWORD'],
|
|
)
|
|
|
|
return g.db_connection
|
|
|
|
def query(query, **params):
|
|
try:
|
|
cursor = db().cursor()
|
|
cursor.paramstyle = "named"
|
|
cursor.execute(query, params)
|
|
db().commit()
|
|
|
|
if cursor._cached_rows:
|
|
columns = [x[0] for x in cursor.description]
|
|
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
# rdef = namedtuple('row', ' '.join([x[0] for x in cursor.description]))
|
|
# rows = list(map(rdef._make, cursor.fetchall()))
|
|
else:
|
|
rows = []
|
|
|
|
return DBResult(
|
|
rows=rows,
|
|
rowcount=cursor.rowcount,
|
|
)
|
|
except:
|
|
db().rollback()
|
|
raise
|