Use SQLAlchemy to connect to DB.

communication
junos 2020-12-24 16:06:23 +01:00
parent 33d048d11b
commit 8d114ebb13
3 changed files with 19 additions and 13 deletions

View File

@ -9,4 +9,5 @@ dependencies:
- flake8 - flake8
- mypy - mypy
- psycopg2 - psycopg2
- python-dotenv - python-dotenv
- sqlalchemy

View File

@ -1,11 +1,17 @@
import os import os
import sqlalchemy.engine.url
from dotenv import load_dotenv from dotenv import load_dotenv
load_dotenv() load_dotenv()
db_host = "212.235.208.113"
db_database = "staw"
db_user = "staw_db"
db_password = os.getenv("DB_PASSWORD") db_password = os.getenv("DB_PASSWORD")
db_uri = sqlalchemy.engine.url.URL(
drivername='postgresql+psycopg2',
username="staw_db",
password=db_password,
host="212.235.208.113",
port=5432,
database="staw"
)

View File

@ -1,19 +1,18 @@
import unittest import unittest
import psycopg2 from sqlalchemy import create_engine
from setup import db_database, db_host, db_password, db_user from setup import db_uri
class DatabaseConnection(unittest.TestCase): class DatabaseConnection(unittest.TestCase):
def setUp(self): def setUp(self):
self.conn = psycopg2.connect( self.engine = create_engine(db_uri, echo=True)
host=db_host, user=db_user, password=db_password, database=db_database,
)
def tearDown(self): def tearDown(self):
self.conn.close() self.engine.dispose()
def test_connection(self): def test_connection(self):
with self.conn.cursor() as cursor: with self.engine.connect() as connection:
self.assertIsNotNone(cursor) self.assertIsNotNone(connection)
connection.close()