Compare commits
3 Commits
997dc064e6
...
83f1dc6a9f
Author | SHA1 | Date | |
---|---|---|---|
83f1dc6a9f | |||
1ef515e265 | |||
e3d8a31285 |
37
database.py
37
database.py
@ -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 os
|
||||||
import psycopg2
|
import psycopg2
|
||||||
|
|
||||||
DB_ADDRESS = os.environ['DB_ADDRESS']
|
|
||||||
DB_USER = os.environ['DB_USER']
|
DB_ADDRESS = os.getenv('DB_ADDRESS', 'localhost')
|
||||||
DB_PASSWORD = os.environ['DB_PASSWORD']
|
DB_PORT = os.getenv('DB_PORT', 5432)
|
||||||
DB_NAME = os.environ['DB_NAME']
|
DB_USER = os.getenv('DB_USER', 'postgres')
|
||||||
|
DB_PASSWORD = os.getenv('DB_PASSWORD', '')
|
||||||
|
DB_NAME = os.getenv('DB_NAME', 'postgres')
|
||||||
|
|
||||||
|
|
||||||
class DBConnector:
|
class DBConnector:
|
||||||
@ -15,16 +20,34 @@ class DBConnector:
|
|||||||
with DBConnector() as db:
|
with DBConnector() as db:
|
||||||
db.read()
|
db.read()
|
||||||
"""
|
"""
|
||||||
def __enter__(self):
|
def _db_start(self):
|
||||||
self.conn = psycopg2.connect()
|
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()
|
self.cur = self.conn.cursor()
|
||||||
|
|
||||||
def __exit__(self):
|
def _db_stop(self):
|
||||||
self.cur.close()
|
self.cur.close()
|
||||||
self.conn.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):
|
def _query(self, sql):
|
||||||
|
try:
|
||||||
self.cur.execute(sql)
|
self.cur.execute(sql)
|
||||||
|
result = self.cur.fetchall()
|
||||||
|
except psycopg2.DatabaseError as e:
|
||||||
|
result = []
|
||||||
|
return result
|
||||||
|
|
||||||
def read(self, **kwargs):
|
def read(self, **kwargs):
|
||||||
"""Read rows from a database that match the specified filters.
|
"""Read rows from a database that match the specified filters.
|
||||||
|
Loading…
x
Reference in New Issue
Block a user