add DB class and fit to tests

pyheatpump/__init__.py

 
 def version():
     return 'PyHeatpump version : {}'.format(__version__)
+
+from .db import DB

pyheatpump/cli.py

 
 from pyheatpump.logger import logger_init
 from pyheatpump.models import *
-from pyheatpump.db import commit
 
 CONTEXT_SETTINGS={
     'default_map':{'run': {}}
                 'value': res[r]})
             val.insert()
 
-    commit()
     logger.info('Successfully read all variables')
 
 
 
     control_data = get_resp.json()
     if h.control(control_data):
-        commit()
         logger.info('GET to supervisor succeded : updated values')
         set_last_update(int(datetime.now().strftime('%s')))
     else:

pyheatpump/db.py

 
 conn = None
 
-def connect():
-    global conn
-    if conn is None:
+class DB():
+    Conn = None
+
+    @staticmethod
+    def connect():
+        if DB.Conn:
+            try:
+                DB.Conn.cursor()
+                return True
+            except sqlite3.ProgrammingError:
+                pass
+            except Exception as e:
+                print(e)
+                sys.exit(1)
+
         logger.info('Will connect to database {}'.format(
-           config['heatpump']['database']))
-        conn = sqlite3.connect(config['heatpump']['database'])
-        conn.row_factory = sqlite3.Row
-    return conn
+            config['heatpump']['database']))
+        DB.Conn = sqlite3.connect(
+            config['heatpump']['database'],
+            isolation_level=None)
+        DB.Conn.row_factory = sqlite3.Row
+        return True
+
+
+    def commit():
+        """
+            Default mode of connecting is autocommit
+        """
+        pass
+
 
-def commit():
-    global conn
-    logger.info('Commit database')
-    conn.commit()
+    @staticmethod
+    def initialize(filename):
+        if DB.Conn:
+            DB.Conn.close()
 
-def initialize(filename):
-    p = Popen(
-        '/usr/bin/env sqlite3 -init {} {}'.format(filename, config['heatpump']['database']),
-        shell=True
-    )
-    return True if p.wait() == 0 else False
+        p = Popen(
+            '/usr/bin/env sqlite3 -init {} {}'.format(filename, config['heatpump']['database']),
+            shell=True)
 
+        return True if p.wait() == 0 else False
 
-def sql(query):
-    if conn is None:
-        logger.debug(f'Connect to the DB\n')
-        connect()
 
-    cursor = conn.cursor()
-    logger.debug(f'Will execute query : \n{query}\n')
-    cursor.execute(query)
+    @staticmethod
+    def sql(query, params={}):
+        print(query)
+        print(params)
+        if not DB.connect():
+            raise Exception('Can\'t connect to DB')
+
+        return DB.Conn.execute(query, params or {})
+
+    @staticmethod
+    def sqlf(filename):
+        with open(filename) as f:
+            if not DB.connect():
+                raise Exception('Can\'t connect to DB')
+
+            try:
+                return DB.Conn.execute((' '.join(f.readlines())).replace('\n', ''))
+            except StopIteration:
+                print('failed')
+                return []
+            except sqlite3.IntegrityError as e:
+                print(e)
+                return []
 
-    conn.close()
-    logger.debug(f'Connection closed\n')
-    return cursor
 
 class RowClass(object):
     def __init__(self, **kwargs):
     def select(self, key, tablename):
         attr = getattr(self, key)
         if type(attr) == str:
-            q = f"SELECT * FROM {tablename} WHERE {key} LIKE '{attr}'"
+            q = ("SELECT * FROM {tablename} WHERE ? LIKE '?'", (key, attr))
         elif type(attr) == int:
-            q = f"SELECT * FROM {tablename} WHERE {key} = {attr}"
+            q = ("SELECT * FROM {tablename} WHERE ? LIKE ?", (key, attr))
+
+        res = []
+        for row in DB.sql(*q):
+            res.append(res)
 
-        return sql(q)
+        return res

pyheatpump/heatpump.py

 from starlette.routing import Router, Route
 from starlette.responses import JSONResponse
 from .config import config
-from pyheatpump.db import sql
 
 from pyheatpump.models import *
 

pyheatpump/models/variable.py

 from pyheatpump.db import RowClass
-from pyheatpump.db import sql
+from pyheatpump.db import DB
 from datetime import date
 from pyheatpump.modbus import rtu
 
     def insert(self):
 
         try:
-            sql(
+            DB.sql(
             """INSERT INTO variable
             (type, address)
             VALUES
-            ('{}', {})
-            """.format(
-                self.type, self.address
-            ))
+            (:type, :address)
+            """, self.__dict__)
             return True
         except Exception as e:
             print(e)
 
     def exists(self):
         try:
-            return bool(next(sql(
+            return bool(next(DB.sql(
             """SELECT 1 FROM variable
-            WHERE type = '{}' AND address = {}
-            """.format(self.type, self.address))))
+            WHERE type=:type AND address=:address
+            """, self.__dict__)))
         except StopIteration:
             return False
 
     def getall_of_type(type: VariableType) -> dict:
         return {
             row['address']: Variable(**dict(row))
-            for row in sql(
+            for row in DB.sql(
             """SELECT * FROM variable
-            WHERE type = '{}'
-                AND address >= {}
-                AND address <= {}""".format(
-            type, type.start_address, type.end_address))
+            WHERE type = :slabel
+                AND address >= :start_address
+                AND address <= :end_address""",
+            type.__dict__)
        }
 
 
     def getall_values_of_type_since(type: VariableType, since: int) -> dict:
         floatcast = lambda x: round(float(x) / 10, 2)
         cast_fct = floatcast if type.slabel == 'A' else lambda x: x
+        type.__dict__.update({'since': since})
 
-        #offset = 10000 if (type.slabel == 'D') else 0
         return {
             row['address']: cast_fct(row['value'])
-            for row in sql(
+            for row in DB.sql(
             """SELECT var.address as address, val.value
             FROM variable var
             LEFT JOIN var_value val ON
                 AND var.address = val.address
                 AND var.last_update = val.time
             WHERE
-                var.type = '{}'
-                AND var.address >= {}
-                AND var.address <= {}
-                AND var.last_update > {}""".format(
-            type, type.start_address, type.end_address, since))
+                var.type = :slabel
+                AND var.address >= :start_address
+                AND var.address <= :end_address
+                AND var.last_update > :since""",
+            type.__dict__)
         }
 
 

pyheatpump/models/variable_type.py

-from pyheatpump.db import RowClass
-from pyheatpump.db import sql
+from pyheatpump.db import DB, RowClass
 
 
 class VariableType(RowClass):
         q = ['UPDATE var_type SET']
         updates = []
         if self.start_address is not None:
-            updates.append(f'start_address = {self.start_address}')
+            updates.append(f'start_address = :start_address')
         if self.end_address is not None:
-            updates.append(f'end_address = {self.end_address}')
+            updates.append(f'end_address = :end_address')
         if len(updates) == 0:
             return
         q.append(','.join(updates))
-        q.append(f"WHERE slabel = '{self.slabel}'")
+        q.append(f"WHERE slabel = :slabel'")
 
-        return sql(' '.join(q))
+        return DB.sql(' '.join(q), self.__dict__)
 
 
     def get_variables(self):
     def getall():
         return {
             row['label']: VariableType(**dict(row))
-            for row in sql('SELECT * FROM var_type')
+            for row in DB.sql('SELECT * FROM var_type')
         }
 
     @staticmethod
 
         try:
             return VariableType(**dict(
-                next(sql(
+                next(DB.sql(
                 """
                 SELECT * FROM var_type
-                WHERE slabel = '{}'
-                """.format(slabel)))))
+                WHERE slabel = ?
+                """, slabel))))
         except StopIteration:
             raise NameError
 

pyheatpump/models/variable_value.py

-from pyheatpump.db import RowClass
-from pyheatpump.db import sql
+from pyheatpump.db import DB, RowClass
 from datetime import datetime
 from pprint import pprint
 
             pass
 
         try:
-            sql(
-            """INSERT INTO var_value
-            (type, address, value)
-            VALUES
-            ('{}', {}, {})""".format(
-                self.type, self.address, self.value
-            ))
+            DB.sql(
+                """INSERT INTO var_value
+                (type, address, value)
+                VALUES
+                (:type, :address, :value)""",
+                self.__dict__
+            )
             return True
         except Exception as e:
             print(e)
     @staticmethod
     def get(type, address, time=datetime.now()):
         try:
-            row = next(sql(
+            row = next(DB.sql(
             """SELECT * FROM var_value
             WHERE
-                type = '{}'
-                AND address = {}
-                AND time <= {}
+                type = :type
+                AND address = :address
+                AND time <= :time
             ORDER BY time DESC
-            LIMIT 1""".format(
-                type, address, int(time.strftime('%s'))+1
-            )))
+            LIMIT 1""", {
+                'type':type, 'address':address, 'time':int(time.strftime('%s'))+1
+            }))
             return VariableValue(**dict(row))
         except StopIteration as e:
             raise e

pyheatpump/variable_types.py

 import json
 
 # pyHeatpump modules
-from pyheatpump.db import sql
 from pyheatpump.models.variable_type import VariableType
 
 

pyheatpump/variable_values.py

 import json
 
 # pyHeatpump modules
-from pyheatpump.db import sql
 from pyheatpump.models.variable import Variable
 from pyheatpump.models.variable_type import VariableType
 from pyheatpump.models.variable_value import VariableValue

pyheatpump/variables.py

 import json
 
 # pyHeatpump modules
-from pyheatpump.db import sql
 from pyheatpump.models.variable import Variable