query_gdbs.py 1015 Bytes
Newer Older
Carl Schaffer's avatar
Carl Schaffer committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import logging
from datetime import datetime, timedelta

import sqlalchemy as sa

_logger = logging.getLogger(__name__)
_gdbs_url = 'postgresql://readuser:logit@dockertest:4321/gregor'
try:
    gdbs_engine = sa.create_engine(_gdbs_url)
    gdbs_engine.connect()
except:
    _logger.warning(
        f'Could not connect to GREGOR database through {_gdbs_url}, are you on the KIS network and is the ssh tunnel running?')
    raise IOError('GDBS is not online')


def query_gdbs(table, datetime, engine=gdbs_engine):
    query = f"""
    SELECT value, timestamp, ttl from {table} where timestamp < '{str(datetime)}' order by timestamp desc limit 1 
    """
    res = engine.execute(query)
    value, timestamp, ttl = res.fetchall()[0]
    age = datetime - timestamp
    is_valid = age < timedelta(seconds=ttl)
    return value, is_valid, age


if __name__ == '__main__':
    engine = gdbs_engine
    print(query_gdbs('xcen', datetime.now(), engine=engine))
    print(query_gdbs('ycen', datetime.now(), engine=engine))