Skip to content
This repository has been archived by the owner on Jul 13, 2022. It is now read-only.

Commit

Permalink
QPT-36798: Create is db available util function (#12)
Browse files Browse the repository at this point in the history
* function plus unit tests

* pre-commit

* Version bumped to 0.13.0

Co-authored-by: ns-circle-ci <[email protected]>
  • Loading branch information
ns-bdesimone and ns-circle-ci authored Oct 5, 2021
1 parent c556398 commit 9f2cb8a
Show file tree
Hide file tree
Showing 4 changed files with 59 additions and 2 deletions.
2 changes: 1 addition & 1 deletion pynocular/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
"""Lightweight ORM that lets you query your database using Pydantic models and asyncio"""

__version__ = "0.12.0"
__version__ = "0.13.0"

from pynocular.engines import DatabaseType, DBInfo
22 changes: 22 additions & 0 deletions pynocular/db_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,28 @@
logger = logging.getLogger()


async def is_database_available(db_info: DBInfo) -> bool:
"""Check if the database is available
Args:
db_info: A database's connection information
Returns:
true if the DB exists
"""
engine = None
try:
engine = await DBEngine.get_engine(db_info)
await engine.acquire()
return True
except Exception:
return False
finally:
if engine:
engine.close()


async def create_new_database(connection_string: str, db_name: str) -> None:
"""Create a new database database for testing
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "pynocular"
version = "0.12.0"
version = "0.13.0"
description = "Lightweight ORM that lets you query your database using Pydantic models and asyncio"
authors = [
"RJ Santana <[email protected]>",
Expand Down
35 changes: 35 additions & 0 deletions tests/unit/test_db_util.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
"""Tests for the db_util module"""
import os

import pytest

from pynocular.db_util import is_database_available
from pynocular.engines import DatabaseType, DBInfo

db_user_password = str(os.environ.get("DB_USER_PASSWORD"))
test_db_name = str(os.environ.get("TEST_DB_NAME", "test_db"))
test_connection_string = str(
os.environ.get(
"TEST_DB_CONNECTION_STRING",
f"postgresql://postgres:{db_user_password}@localhost:5432/{test_db_name}?sslmode=disable",
)
)
test_db = DBInfo(DatabaseType.aiopg_engine, test_connection_string)


class TestDBUtil:
"""Test cases for DB util functions"""

@pytest.mark.asyncio
async def test_is_database_available(self) -> None:
"""Test successful database connection"""
available = await is_database_available(test_db)
assert available is True

@pytest.mark.asyncio
async def test_is_database_not_available(self) -> None:
"""Test db connection unavailable"""
invalid_connection_string = f"postgresql://postgres:{db_user_password}@localhost:5432/INVALID?sslmode=disable"
non_existing_db = DBInfo(DatabaseType.aiopg_engine, invalid_connection_string)
available = await is_database_available(non_existing_db)
assert available is False

0 comments on commit 9f2cb8a

Please sign in to comment.