Abstract model for support both DBMS

model/__init__.py

@@ -1,4 +1,7 @@
-import model.postgres_model
+from model.postgres_model import PostgresModel
+from model.sqlite_model import Sqlite3Model
+from model.abstract_model import AbstractModel
+
 import utils
 from EDMCLogging import get_main_logger
 import os
@@ -7,17 +10,21 @@ logger = get_main_logger()
 
 env_choose = os.getenv('DB_NAME')
 
+model: AbstractModel
+
 if env_choose == 'postgres':
     logger.info('Using postgres DB')
-    from .import postgres_model as model
+    model = PostgresModel()
 
 elif env_choose == 'sqlite':
     logger.info('Using sqlite DB')
-    from .import sqlite_model as model
+    model = Sqlite3Model()
 
 else:
     logger.info('Using sqlite DB')
-    from . import sqlite_model as model
+    model = Sqlite3Model()
+
+model.open_model()
 
 model.get_diff_action_id = utils.measure(model.get_diff_action_id)
 model.get_activity_changes = utils.measure(model.get_activity_changes)

model/abstract_model.py

@@ -0,0 +1,25 @@
+import abc
+
+
+class AbstractModel(abc.ABC):
+
+    @abc.abstractmethod
+    def open_model(self) -> None:
+        raise NotImplemented
+
+    @abc.abstractmethod
+    def close_model(self) -> None:
+        raise NotImplemented
+
+    @abc.abstractmethod
+    def get_activity_changes(self, platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp)\
+            -> list:
+        raise NotImplemented
+
+    @abc.abstractmethod
+    def insert_leaderboard_db(self, leaderboard_list: dict) -> None:
+        raise NotImplemented
+
+    @abc.abstractmethod
+    def get_diff_action_id(self, action_id: int) -> list:
+        raise NotImplemented

model/postgres_model.py

@@ -1,6 +1,5 @@
 import json
 import os
-import sqlite3
 import typing
 
 import psycopg2.extensions
@@ -8,113 +7,119 @@ import psycopg2.extras
 
 from .sqlite_cache import cache
 from . import postgres_sql_requests
+from .abstract_model import AbstractModel
 import utils
 from EDMCLogging import get_main_logger
 
 logger = get_main_logger()
 logger.propagate = False
 
-db: psycopg2.extensions.connection = psycopg2.connect(
-    user=os.environ['DB_USERNAME'],  # user2
-    password=os.environ['DB_PASSWORD'],  # 1
-    host=os.environ['DB_HOSTNAME'],  # 192.168.1.68
-    port=os.environ['DB_PORT'],  # 5432
-    database=os.environ['DB_DATABASE'],  # test0
-    cursor_factory=psycopg2.extras.DictCursor)
 
-with db:
-    with db.cursor() as cursor:
-        cursor.execute(postgres_sql_requests.schema_create)  # schema creation
+class PostgresModel(AbstractModel):
+    db: psycopg2.extensions.connection
 
+    def open_model(self):
+        self.db: psycopg2.extensions.connection = psycopg2.connect(
+            user=os.environ['DB_USERNAME'],  # user2
+            password=os.environ['DB_PASSWORD'],  # 1
+            host=os.environ['DB_HOSTNAME'],  # 192.168.1.68
+            port=os.environ['DB_PORT'],  # 5432
+            database=os.environ['DB_DATABASE'],  # test0
+            cursor_factory=psycopg2.extras.DictCursor)
 
-def get_activity_changes(platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp) -> list:
-    cache_key: str = f'{platform}_{leaderboard_type}_{limit}_{low_timestamp}_{high_timestamp}'
-    cached_result: typing.Union[str, None] = cache.get(cache_key)
+        with self.db:
+            with self.db.cursor() as cursor:
+                cursor.execute(postgres_sql_requests.schema_create)  # schema creation
 
-    if cached_result is not None:
-        logger.debug(f'Cached result for {cache_key}')
-        return json.loads(cached_result)
+    def close_model(self):
+        self.db.close()
 
-    logger.debug(f'Not cached result for {cache_key}')
+    def get_activity_changes(self, platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp)\
+            -> list:
+        cache_key: str = f'{platform}_{leaderboard_type}_{limit}_{low_timestamp}_{high_timestamp}'
+        cached_result: typing.Union[str, None] = cache.get(cache_key)
 
-    with db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
-        cursor.execute(postgres_sql_requests.select_activity_pretty_names, {
-            'LB_type': utils.LeaderboardTypes(leaderboard_type.lower()).value,
-            'platform': utils.Platform(platform.upper()).value,
-            'limit': limit,
-            'high_timestamp': high_timestamp,
-            'low_timestamp': low_timestamp
-        })
+        if cached_result is not None:
+            logger.debug(f'Cached result for {cache_key}')
+            return json.loads(cached_result)
 
-        result: list = cursor.fetchall()
+        logger.debug(f'Not cached result for {cache_key}')
 
-    if not cache.disabled:
-        cache.set(cache_key, json.dumps(result))
+        with self.db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
+            cursor.execute(postgres_sql_requests.select_activity_pretty_names, {
+                'LB_type': utils.LeaderboardTypes(leaderboard_type.lower()).value,
+                'platform': utils.Platform(platform.upper()).value,
+                'limit': limit,
+                'high_timestamp': high_timestamp,
+                'low_timestamp': low_timestamp
+            })
 
-    return result
+            result: list = cursor.fetchall()
 
+        if not cache.disabled:
+            cache.set(cache_key, json.dumps(result))
 
-def insert_leaderboard_db(leaderboard_list: dict) -> None:
-    """
-    Takes leaderboard as list, it platform, type, db connection and insert leaderboard to DB
+        return result
 
-    :param leaderboard_list: list from request_leaderboard
-    :return:
-    """
+    def insert_leaderboard_db(self, leaderboard_list: dict) -> None:
+        """
+        Takes leaderboard as list, it platform, type, db connection and insert leaderboard to DB
 
-    platform: str = leaderboard_list['platform']
-    LB_type: str = leaderboard_list['type']
-    leaderboard: list = leaderboard_list['leaderboard']
+        :param leaderboard_list: list from request_leaderboard
+        :return:
+        """
 
-    action_id: int  # not last, current that we will use
+        platform: str = leaderboard_list['platform']
+        LB_type: str = leaderboard_list['type']
+        leaderboard: list = leaderboard_list['leaderboard']
 
-    with db.cursor() as cursor:
-        cursor.execute(postgres_sql_requests.select_last_action_id)
-        action_id_fetch_one: typing.Union[None, dict[str, int]] = cursor.fetchone()
+        action_id: int  # not last, current that we will use
 
-    if action_id_fetch_one is None:
-        # i.e. first launch
-        action_id = 1  # yep, not 0
+        with self.db.cursor() as cursor:
+            cursor.execute(postgres_sql_requests.select_last_action_id)
+            action_id_fetch_one: typing.Union[None, dict[str, int]] = cursor.fetchone()
 
-    else:
-        action_id = action_id_fetch_one['action_id'] + 1
+        if action_id_fetch_one is None:
+            # i.e. first launch
+            action_id = 1  # yep, not 0
 
-    # Patch for additional values
-    for squad in leaderboard:
-        squad.update({'action_id': action_id, 'LB_type': LB_type, 'platform': platform})
+        else:
+            action_id = action_id_fetch_one['action_id'] + 1
 
-    with db:
-        with db.cursor() as cursor:
-            cursor.executemany(
-                postgres_sql_requests.insert_leader_board,
-                leaderboard)
+        # Patch for additional values
+        for squad in leaderboard:
+            squad.update({'action_id': action_id, 'LB_type': LB_type, 'platform': platform})
 
-    cache.delete_all()  # drop cache
+        with self.db:
+            with self.db.cursor() as cursor:
+                cursor.executemany(
+                    postgres_sql_requests.insert_leader_board,
+                    leaderboard)
 
+        cache.delete_all()  # drop cache
 
-@utils.measure
-def get_diff_action_id(action_id: int) -> list:
-    """
-    Takes action_id and returns which squadrons has been changed in leaderboard as in action_id and
-    experience they got in compassion to action_id - 1 for the same leaderboard and platform
+    def get_diff_action_id(self, action_id: int) -> list:
+        """
+        Takes action_id and returns which squadrons has been changed in leaderboard as in action_id and
+        experience they got in compassion to action_id - 1 for the same leaderboard and platform
 
-    :param action_id:
-    :return:
-    """
-    cache_key: str = f'{action_id}'
-    cached_result: typing.Union[str, None] = cache.get(cache_key)
+        :param action_id:
+        :return:
+        """
+        cache_key: str = f'{action_id}'
+        cached_result: typing.Union[str, None] = cache.get(cache_key)
 
-    if cached_result is not None:
-        logger.debug(f'Cached result for {cache_key}')
-        return json.loads(cached_result)
+        if cached_result is not None:
+            logger.debug(f'Cached result for {cache_key}')
+            return json.loads(cached_result)
 
-    logger.debug(f'Not cached result for {cache_key}')
+        logger.debug(f'Not cached result for {cache_key}')
 
-    with db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
-        cursor.execute(postgres_sql_requests.select_diff_by_action_id, {'action_id': action_id})
-        result: list = cursor.fetchall()
+        with self.db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cursor:
+            cursor.execute(postgres_sql_requests.select_diff_by_action_id, {'action_id': action_id})
+            result: list = cursor.fetchall()
 
-    if not cache.disabled:
-        cache.set(cache_key, json.dumps(result))
+        if not cache.disabled:
+            cache.set(cache_key, json.dumps(result))
 
-    return result
+        return result

model/sqlite_model.py

@@ -3,6 +3,7 @@ import typing
 import json
 
 from . import sqlite_sql_requests
+from .abstract_model import AbstractModel
 from .sqlite_cache import cache
 import utils
 from EDMCLogging import get_main_logger
@@ -10,93 +11,100 @@ from EDMCLogging import get_main_logger
 logger = get_main_logger()
 logger.propagate = False
 
-db: sqlite3.Connection = sqlite3.connect('squads_stat.sqlite3', check_same_thread=False)
 
-db.executescript(sqlite_sql_requests.schema_create)  # schema creation
+class Sqlite3Model(AbstractModel):
+    db: sqlite3.Connection
 
-# thx https://stackoverflow.com/a/48789604
-db.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))
+    def open_model(self):
+        self.db = sqlite3.connect('squads_stat.sqlite3', check_same_thread=False)
 
+        self.db.executescript(sqlite_sql_requests.schema_create)  # schema creation
 
-def get_activity_changes(platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp) -> list:
-    cache_key: str = f'{platform}_{leaderboard_type}_{limit}_{low_timestamp}_{high_timestamp}'
-    cached_result: typing.Union[str, None] = cache.get(cache_key)
+        # thx https://stackoverflow.com/a/48789604
+        self.db.row_factory = lambda c, r: dict(zip([col[0] for col in c.description], r))
 
-    if cached_result is not None:
-        logger.debug(f'Cached result for {cache_key}')
-        return json.loads(cached_result)
+    def close_model(self):
+        self.db.close()
 
-    logger.debug(f'Not cached result for {cache_key}')
+    def get_activity_changes(self, platform: str, leaderboard_type: str, limit: int, low_timestamp, high_timestamp)\
+            -> list:
+        cache_key: str = f'{platform}_{leaderboard_type}_{limit}_{low_timestamp}_{high_timestamp}'
+        cached_result: typing.Union[str, None] = cache.get(cache_key)
 
-    sql_req: sqlite3.Cursor = db.execute(sqlite_sql_requests.select_activity_pretty_names, {
-        'LB_type': utils.LeaderboardTypes(leaderboard_type.lower()).value,
-        'platform': utils.Platform(platform.upper()).value,
-        'limit': limit,
-        'high_timestamp': high_timestamp,
-        'low_timestamp': low_timestamp
-    })
+        if cached_result is not None:
+            logger.debug(f'Cached result for {cache_key}')
+            return json.loads(cached_result)
 
-    result: list = sql_req.fetchall()
+        logger.debug(f'Not cached result for {cache_key}')
 
-    if not cache.disabled:
-        cache.set(cache_key, json.dumps(result))
+        sql_req: sqlite3.Cursor = self.db.execute(sqlite_sql_requests.select_activity_pretty_names, {
+            'LB_type': utils.LeaderboardTypes(leaderboard_type.lower()).value,
+            'platform': utils.Platform(platform.upper()).value,
+            'limit': limit,
+            'high_timestamp': high_timestamp,
+            'low_timestamp': low_timestamp
+        })
 
-    return result
+        result: list = sql_req.fetchall()
 
+        if not cache.disabled:
+            cache.set(cache_key, json.dumps(result))
 
-def insert_leaderboard_db(leaderboard_list: dict) -> None:
-    """
-    Takes leaderboard as list, it platform, type, db connection and insert leaderboard to DB
-    :param leaderboard_list: list from request_leaderboard
-    :return:
-    """
+        return result
 
-    platform: str = leaderboard_list['platform']
-    LB_type: str = leaderboard_list['type']
-    leaderboard: list = leaderboard_list['leaderboard']
+    def insert_leaderboard_db(self, leaderboard_list: dict) -> None:
+        """
+        Takes leaderboard as list, it platform, type, db connection and insert leaderboard to DB
+        :param leaderboard_list: list from request_leaderboard
+        :return:
+        """
 
-    action_id: int  # not last, current that we will use
+        platform: str = leaderboard_list['platform']
+        LB_type: str = leaderboard_list['type']
+        leaderboard: list = leaderboard_list['leaderboard']
 
-    sql_req_action_id: sqlite3.Cursor = db.execute(sqlite_sql_requests.select_last_action_id)
-    action_id_fetch_one: typing.Union[None, dict[str, int]] = sql_req_action_id.fetchone()
-    if action_id_fetch_one is None:
-        # i.e. first launch
-        action_id = 1  # yep, not 0
+        action_id: int  # not last, current that we will use
 
-    else:
-        action_id = action_id_fetch_one['action_id'] + 1
+        sql_req_action_id: sqlite3.Cursor = self.db.execute(sqlite_sql_requests.select_last_action_id)
+        action_id_fetch_one: typing.Union[None, dict[str, int]] = sql_req_action_id.fetchone()
+        if action_id_fetch_one is None:
+            # i.e. first launch
+            action_id = 1  # yep, not 0
 
-    # Patch for additional values
-    for squad in leaderboard:
-        squad.update({'action_id': action_id, 'LB_type': LB_type, 'platform': platform})
+        else:
+            action_id = action_id_fetch_one['action_id'] + 1
 
-    with db:
-        db.executemany(
-            sqlite_sql_requests.insert_leader_board,
-            leaderboard)
+        # Patch for additional values
+        for squad in leaderboard:
+            squad.update({'action_id': action_id, 'LB_type': LB_type, 'platform': platform})
 
-    cache.delete_all()  # drop cache
+        with self.db:
+            self.db.executemany(
+                sqlite_sql_requests.insert_leader_board,
+                leaderboard)
 
+        cache.delete_all()  # drop cache
 
-def get_diff_action_id(action_id: int) -> list:
-    """
-    Takes action_id and returns which squadrons has been changed in leaderboard as in action_id and
-    experience they got in compassion to action_id - 1 for the same leaderboard and platform
-    :param action_id:
-    :return:
-    """
-    cache_key: str = f'{action_id}'
-    cached_result: typing.Union[str, None] = cache.get(cache_key)
+    def get_diff_action_id(self, action_id: int) -> list:
+        """
+        Takes action_id and returns which squadrons has been changed in leaderboard as in action_id and
+        experience they got in compassion to action_id - 1 for the same leaderboard and platform
+        :param action_id:
+        :return:
+        """
+        cache_key: str = f'{action_id}'
+        cached_result: typing.Union[str, None] = cache.get(cache_key)
 
-    if cached_result is not None:
-        logger.debug(f'Cached result for {cache_key}')
-        return json.loads(cached_result)
+        if cached_result is not None:
+            logger.debug(f'Cached result for {cache_key}')
+            return json.loads(cached_result)
 
-    logger.debug(f'Not cached result for {cache_key}')
-    sql_req: sqlite3.Cursor = db.execute(sqlite_sql_requests.select_diff_by_action_id, {'action_id': action_id})
-    result: list = sql_req.fetchall()
+        logger.debug(f'Not cached result for {cache_key}')
+        sql_req: sqlite3.Cursor = self.db.execute(sqlite_sql_requests.select_diff_by_action_id,
+                                                  {'action_id': action_id})
+        result: list = sql_req.fetchall()
 
-    if not cache.disabled:
-        cache.set(cache_key, json.dumps(result))
+        if not cache.disabled:
+            cache.set(cache_key, json.dumps(result))
 
-    return result
+        return result