Commit 7c8831e1 authored by Oriol Guitart's avatar Oriol Guitart
Browse files

Module database added with unit tests included. Only general_properties table fully working

parent 904734bb
Pipeline #318 failed with stage
in 49 seconds
from .database import Database
from .general_prop import GeneralProp
from chemicalchecker.util import Config
from chemicalchecker.util import logged
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import NullPool
import psycopg2
Base = declarative_base()
config = Config()
def get_connection():
datab = psycopg2.connect(host=config.DB.host, user=config.DB.username,
password=config.DB.password, database=config.DB.main_DB)
return datab
def get_engine():
if config.DB.dialect == 'sqlite':
con = config.DB.dialect + ':///' + config.DB.file
else:
con = config.DB.dialect + '://' + config.DB.username + ':' + config.DB.password + \
'@' + config.DB.host + ':' + config.DB.port + '/' + config.DB.main_DB
#engine = create_engine('sqlite:///orm_in_detail.sqlite', echo=True)
engine = create_engine(con, echo=True, poolclass=NullPool)
return engine
def get_session():
Session = sessionmaker()
engine = get_engine()
Session.configure(bind=engine)
session = Session()
return session
@logged
class Database(object):
"""A class to query the cc database bla bla."""
Base = None
def __init__(self):
self.__log.debug('Database class')
from chemicalchecker.util import logged
from .database import Base, get_session, get_engine
from sqlalchemy import Column, Integer, Text, Float
@logged
class GeneralProp(Base):
"""A Signature bla bla."""
__tablename__ = 'physchem'
inchikey = Column(Text, primary_key=True)
mw = Column(Integer)
heavy = Column(Integer)
hetero = Column(Integer)
rings = Column(Integer)
ringaliph = Column(Integer)
ringarom = Column(Integer)
alogp = Column(Float)
mr = Column(Float)
hba = Column(Integer)
hbd = Column(Integer)
psa = Column(Float)
rotb = Column(Integer)
alerts_qed = Column(Integer)
alerts_chembl = Column(Integer)
ro5 = Column(Integer)
ro3 = Column(Integer)
qed = Column(Float)
@staticmethod
def add(kwargs):
""" Method to add a new row to the table.
Args:
kwargs(dict):The data in dictionary format .
"""
GeneralProp.__log.debug(type(kwargs))
if type(kwargs) is dict:
prop = GeneralProp(**kwargs)
GeneralProp.__log.debug(prop.inchikey)
session = get_session()
session.add(prop)
session.commit()
session.close()
@staticmethod
def add_bulk(data, chunk=1000):
""" Method to add a new row to the table.
The data will be in array format. the order is very important
"""
engine = get_engine()
for pos in xrange(0, len(data), chunk):
engine.execute(
GeneralProp.__table__.insert(),
[{"inchikey": row[0], "mw": row[1], "heavy": row[2], "hetero": row[4],
"rings": row[4], "ringaliph": row[5], "ringarom": row[6], "alogp": row[7],
"mr": row[8], "hba": row[9], "hbd": row[10], "psa": row[11], "rotb": row[12],
"alerts_qed": row[13], "alerts_chembl": row[14], "ro5": row[15], "ro3": row[16], "qed": row[17]}
for row in data[pos:pos + chunk]]
)
@staticmethod
def get(key):
""" Method to query general_properties table.
"""
session = get_session()
query = session.query(GeneralProp).filter_by(inchikey=key)
res = query.one_or_none()
session.close()
return res
@staticmethod
def create_table():
engine = get_engine()
Base.metadata.create_all(engine)
......@@ -8,5 +8,6 @@ Sphinx==1.8.1
twine==1.12.1
pytest==3.8.2
pytest-runner==4.2
sqlalchemy
autologging==1.1.0
\ No newline at end of file
{
"PATH": {
"CC_ROOT": "'/aloy/web_checker/'"
},
"DB": {
"dialect": "'postgresql'",
"port": "'5432'",
"host": "'aloy-dbsrv'",
"username": "'sbnb-adm'",
"password": "'db_adm'",
"main_DB": "'mosaic'"
}
}
{
"PATH": {
"CC_ROOT": "'/aloy/web_checker/'"
},
"DB": {
"dialect": "'sqlite'",
"file": "'test.sqlite'"
}
}
import os
import unittest
from chemicalchecker.util import Config
class TestDatabase(unittest.TestCase):
def setUp(self):
# path for test data
test_dir = os.path.dirname(os.path.realpath(__file__))
self.data_dir = os.path.join(test_dir, 'data')
conffile = os.path.join(self.data_dir, 'database_config.json')
Config(conffile)
from chemicalchecker.database import GeneralProp
self.GeneralProp = GeneralProp
GeneralProp.create_table()
def tearDown(self):
test_dir = os.path.dirname(os.path.realpath(__file__))
testfile = os.path.join(test_dir, 'test.sqlite')
os.remove(testfile)
def test_add_bulk(self):
self.GeneralProp.add_bulk([["test1", 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17],
["test2", 1, 22, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17]])
res = self.GeneralProp.get('test1')
self.assertTrue(hasattr(res, 'mw'))
self.assertTrue(res.heavy == 2)
res = self.GeneralProp.get('test2')
self.assertTrue(hasattr(res, 'mw'))
self.assertTrue(res.heavy == 22)
\ No newline at end of file
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment