Compare commits

...

3 Commits

View File

@ -1,10 +1,15 @@
"""This module contains functionality for interacting with a PostgreSQL database. It will automatically handle error
conditions (i.e. missing columns) without terminating the entire program. Use the :py:class:`DBConnector` class to
handle database interactions, either as a standalone object or in a context manager."""
import os
import psycopg2
DB_ADDRESS = os.environ['DB_ADDRESS']
DB_USER = os.environ['DB_USER']
DB_PASSWORD = os.environ['DB_PASSWORD']
DB_NAME = os.environ['DB_NAME']
DB_ADDRESS = os.getenv('DB_ADDRESS', 'localhost')
DB_PORT = os.getenv('DB_PORT', 5432)
DB_USER = os.getenv('DB_USER', 'postgres')
DB_PASSWORD = os.getenv('DB_PASSWORD', '')
DB_NAME = os.getenv('DB_NAME', 'postgres')
class DBConnector:
@ -15,16 +20,34 @@ class DBConnector:
with DBConnector() as db:
db.read()
"""
def __enter__(self):
self.conn = psycopg2.connect()
def _db_start(self):
self.conn = psycopg2.connect(
f"host={DB_ADDRESS} port={DB_PORT} dbname={DB_NAME} user={DB_USER} password={DB_PASSWORD}")
self.cur = self.conn.cursor()
def __exit__(self):
def _db_stop(self):
self.cur.close()
self.conn.close()
def __init__(self):
self._db_start()
def __del__(self):
self._db_stop()
def __enter__(self):
self._db_start()
def __exit__(self):
self._db_stop()
def _query(self, sql):
self.cur.execute(sql)
try:
self.cur.execute(sql)
result = self.cur.fetchall()
except psycopg2.DatabaseError as e:
result = []
return result
def read(self, **kwargs):
"""Read rows from a database that match the specified filters.