Выделение БД и кэша как отдельные модули

App/Waybill/cache/__init__.py

@@ -1,42 +0,0 @@
-import json
-from redis import Redis, ConnectionError, DataError
-
-class RedisCache:
-    def __init__(self, config: dict):
-        self.config = config
-        self.conn = self._connect()
-
-    def _connect(self):
-        try:
-            conn = Redis(**self.config)
-            return conn
-        except DataError as err:
-            print(err)
-            return None
-
-    def set_value(self, name: str, value_dict: dict, ttl: int):
-        try:
-            value_js = json.dumps(value_dict)
-            self.conn.set(name=name, value=value_js)
-            if ttl > 0:
-                self.conn.expire(name, ttl)
-                return True
-        except ConnectionError as err:
-            print(err)
-            return False
-
-    def get_value(self, name: str):
-        try:
-            value_js = self.conn.get(name)
-        except:
-            return None
-        value_js = self.conn.get(name)
-        if value_js:
-            value_dict = json.loads(value_js)
-            return value_dict
-        else:
-            return None
-    
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.conn.close()
-        return True

App/Waybill/cache/wrapper.py

@@ -1,26 +0,0 @@
-from functools import wraps
-from Waybill.cache import RedisCache
-
-def fetch_from_cache(cache_name: str, cache_config: dict):
-    cache_conn = RedisCache(cache_config['redis'])
-    ttl = cache_config['ttl']
-    def decorator(f):
-        # как оно работает
-        # лезем в кэш и смотрим, есть ли там что-то
-        # если есть, то возвращаем кэшированную информацию
-        # если нет, то запускаем декорируемую функцию
-        # достаём оттуда информацию
-        # заносим её в кэш
-        # возвращаем кэшированную информацию
-        @wraps(f)
-        def wrapper(*args, **kwargs):
-            cached_value = cache_conn.get_value(cache_name)
-            print("cached_value=", cached_value)
-            if cached_value:
-                return cached_value
-            response = f(*args, **kwargs)
-            print("response=", response)
-            cache_conn.set_value(cache_name,response,ttl)
-            return response
-        return wrapper
-    return decorator

App/Waybill/db/DBconnect.py

@@ -1,34 +0,0 @@
-import pymysql
-from pymysql.err import *
-class DBContextManager:
-    def __init__(self, db_config : dict):
-        self.db_config = db_config
-        self.connection = None
-        self.cursor = None
-
-    def __enter__(self):
-        try:
-            self.connection = pymysql.connect(
-                host=self.db_config['host'],
-                port=self.db_config['port'],
-                user=self.db_config['user'],
-                password=self.db_config['password'],
-                db=self.db_config['db']
-            )
-            self.cursor = self.connection.cursor()
-            return self.cursor
-        except (OperationalError, KeyError) as err:
-            print(err.args)
-            return None
-    
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        if self.connection and self.cursor:
-            if exc_type:
-                print(exc_type, '\n', exc_val)
-                self.connection.rollback()
-            else:
-                self.connection.commit()
-            self.cursor.close()
-            self.connection.close()
-        return True
-    

App/Waybill/db/sql_provider.py

@@ -1,14 +0,0 @@
-import os
-from string import Template
-
-class SQLProvider:
-    def __init__(self, file_path):
-        self.scripts = {}
-        for file in os.listdir(file_path):
-            _sql = open(f'{file_path}/{file}').read()
-            self.scripts[file] = Template(_sql)
-
-    def get(self, name, params) -> dict:
-        if name not in self.scripts:
-            raise ValueError(f'SQL template {name} not found')
-        return self.scripts[name].substitute(**params)

App/Waybill/db/work.py

@@ -1,16 +0,0 @@
-from .DBconnect import DBContextManager
-
-def select_list(db_config, sql) -> list:
-    with DBContextManager(db_config) as cursor:
-        if cursor is None:
-            raise ValueError("Cursor not created")
-        else:
-            cursor.execute(sql)
-            result = cursor.fetchall()
-            schema = [item[0] for item in cursor.description]
-            lst = [dict(zip(schema, row)) for row in result]
-            return lst
-
-def transaction(cursor, sql):
-    cursor.execute(sql)
-    return True