Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • fejao/c3buttons
1 result
Show changes
Showing
with 1200 additions and 0 deletions
#!/usr/bin/env python3
"""
utils.configuration.from_file.py file for the c3 buttons.
This is the super class used from the other classes that fetches configurations from a file.
"""
import sys
from abc import ABC, abstractmethod
### LOCAL
try:
from utils.configuration.app_logger import set_logger_from_config
except ImportError as e:
print("Unable to import 'set_logger_from_dict' from 'utils.configuration.app_logger.py' file at 'utils.boxes.box_pin.py' file" )
sys.exit(1)
class BoxPin(ABC):
""" TODO
"""
def __init__(self, name: str, configuration: object, box_name: str) -> None:
self.configuration = configuration
self.logger = set_logger_from_config(name=name, config=self.configuration)
self.get_box_config(box_name)
def get_box_config(self, box_name):
self.logger.debug(f"get_box_config - box_name: {box_name}")
self.box_name = box_name
box_info = self.configuration.boxes.get(box_name)
self.pin_button = box_info.get('pin_button')
self.pin_led = box_info.get('pin_led')
self.enabled_led=self.configuration.box_led_enabled
@abstractmethod
def set_pin_usage(self):
pass
@abstractmethod
def activated(self):
pass
#!/usr/bin/env python3
"""
utils.configuration.app_configuration.py file for the c3buttons system.
In this file the configuration will be setted from this order:
config.yml -> environment vars
"""
###
### DEPENDENCIES
###
import sys
### LOCAL
try:
from .default_values import DEFAULT_VALUES
except ImportError as e:
print("Unable to import 'utils.configuration.default_vals.py' from 'utils.configuration.app_configuration.py' file")
sys.exit(1)
try:
from .app_logger import set_logger_from_default
except ImportError as e:
print("Unable to import 'setup_logger' from 'utils.configuration.app_logger.py' file at the 'utils.configuration.app_configuration.py'")
sys.exit(1)
try:
from .from_file_environment import ConfigurationFromDotEnv
except ImportError as e:
print("Unable to import 'ConfigurationFromFile' from 'utils.configuration.from_file_environment.py' file at the 'utils.configuration.app_configuration.py'")
sys.exit(1)
try:
from .from_file_yaml import ConfigurationFromYamlFile
except ImportError as e:
print("Unable to import 'ConfigurationFromYamlFile' from 'utils.configuration.from_file_yaml.py' file at the 'utils.configuration.app_configuration.py'")
sys.exit(1)
class AppConfiguration():
"""
Class for setting the configuration of the app.
"""
def __init__(self) -> None:
"""
Function for initialiazing the AppConfiguration class.
Sets the class variables.
Returns
-------
None
"""
self.logger = set_logger_from_default(name=__name__, values_default=DEFAULT_VALUES)
self.logger.info("---> STARTED")
self.__configuration_colors__()
self.set_attr_from_parsed()
if self.debug:
self.__debug_attrs__()
def check_missing_attrs(self, from_env: dict, from_yaml: dict) -> dict:
self.logger.debug("check_missin_attrs")
ret_dict = {}
# If in from_env and not in from_yaml
for name, value in from_env.dict_values.items():
if name not in from_yaml.dict_values.keys():
ret_dict[name] = value
# if name in from_yaml.dict_values.keys():
else:
if name == "boxes":
yaml_value = from_yaml.dict_values.get("boxes")
if len(value) >= len(yaml_value):
boxes_dict = value
boxes_list = from_env.dict_values.get("boxes_used")
self.logger.info("---> Using boxes configuration from OS environments")
else:
boxes_dict = yaml_value
boxes_list = from_yaml.dict_values.get("boxes_used")
self.logger.info("---> Using boxes configuration from config.yml file")
# ret_dict[name] = new_value
ret_dict["boxes"] = boxes_dict
ret_dict["boxes_used"] = boxes_list
elif name == "boxes_used":
pass
else:
ret_dict[name] = value
### DEBUG
# from pprint import pp
# print(f"name: {name} | ret_dict: {ret_dict}")
# If in from_yaml and not in from_env
for name, value in from_yaml.dict_values.items():
if name not in ret_dict.keys():
ret_dict[name] = value
# from pprint import pp
# import pdb; pdb.set_trace()
# pp(ret_dict)
return ret_dict
def attr_set_from_dict(self, dict_values: dict) -> None:
self.logger.debug("attr_set_from_dict")
for name, value in dict_values.items():
setattr(self, name.lower(), value)
def set_attr_from_parsed(self) -> None:
self.logger.debug("set_attr_from_parsed")
### Get attrs from env
from_env = ConfigurationFromDotEnv(default_values= DEFAULT_VALUES)
### If not using, set it from the os.env
if not DEFAULT_VALUES.get("DEFAULT_USING_PARSED_YAML"):
dict_values = from_env.dict_values
else:
### Fetch params from YAML file
from_yaml = ConfigurationFromYamlFile(default_values= DEFAULT_VALUES)
dict_values = self.check_missing_attrs(from_env=from_env, from_yaml=from_yaml)
### Set the attributes as a dict aswell
self.attr_set_from_dict(dict_values)
self.dict_values = dict_values
def __configuration_colors__(self) -> None:
"""
Method for setting the terminal output with colors.
Returns
-------
None
"""
self.logger.debug("__configuration_colors__")
self.color_gray = "\033[90m"
self.color_red = "\033[91m"
self.color_green = "\033[92m"
self.color_orange = "\033[93m"
self.color_blue = "\033[94m"
self.color_pink = "\033[95m"
self.color_yellow = "\033[93m"
self.color_end = "\033[0m"
def __debug_attrs__(self) -> None:
"""
Function for printing the configuration parameters from the app.
Returns
-------
None
"""
self.logger.debug("__debug_attrs__")
### PRINT ARGS
print("\n" + self.color_blue + "CONFIGURATION ATTRS:" + self.color_end)
for name, value in self.dict_values.items():
print(self.color_green + f"\t- {name}: " + self.color_end + f"'{value}'")
print("\n")
#!/usr/bin/env python3
"""
utils.configuration.app_logger.py file for the c3 printing system.
"""
###
### DEPENDENCIES
###
import sys
import logging
def setup_logger(name: str,
level: str | None = "INFO",
log_formatter: str | None = None,
log_file: str | None = None,
log_file_append: bool | None = False
) -> logging.Logger:
if level == "DEBUG":
lvl = logging.DEBUG
elif level == "INFO":
lvl = logging.INFO
elif level == "WARNING":
lvl = logging.WARNING
elif level == "ERROR":
lvl = logging.ERROR
else:
print(f"\nERROR ---> Could not set logger for: {name}.\n LogLevel parsed '{level}' is unknown...\n\n")
sys.exit()
# Create a logger
logger = logging.getLogger(name)
# Set the logging level
logger.setLevel(lvl)
# Avoid adding multiple handlers
if not logger.hasHandlers():
# Set log formatter
if log_formatter is None:
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
else:
formatter = logging.Formatter(log_formatter)
if log_file is not None:
if log_file_append:
fh = logging.FileHandler(log_file, mode="a")
else:
fh = logging.FileHandler(log_file)
fh.setLevel(lvl)
fh.setFormatter(formatter)
logger.addHandler(fh)
# Create a console handler (optional, if you still want console output)
ch = logging.StreamHandler()
ch.setLevel(lvl)
# # Change to WARNING to only see warnings/errors in the terminal
# ch.setLevel(logging.WARNING)
# ch.setLevel(logging.INFO)
# Add formatter to both handlers
ch.setFormatter(formatter)
# Add handlers to the logger
logger.addHandler(ch)
return logger
def set_logger_from_default(name: str, values_default: dict) -> logging.Logger:
log_file = None
if values_default.get('DEFAULT_LOG_TO_FILE'):
log_file = values_default.get('DEFAULT_LOG_FILE_PATH')
if values_default.get('DEFAULT_DEBUG'):
logger = setup_logger(
name=name,
level="DEBUG",
log_formatter=values_default.get('DEFAULT_LOG_FORMAT'),
log_file=log_file,
)
else:
logger = setup_logger(
name=name,
# level=logging.DEBUG,
log_formatter=values_default.get('DEFAULT_LOG_FORMAT'),
log_file=log_file,
)
return logger
def set_logger_from_config(name: str, config: object) -> logging.Logger:
log_file = None
log_file_append = True
if config.log_to_file:
log_file = config.log_file_path
log_file_append = config.log_file_append_mode
if config.debug:
logger = setup_logger(
name=name,
level="DEBUG",
log_formatter=config.log_format,
log_file=log_file,
log_file_append=log_file_append,
)
else:
logger = setup_logger(
name=name,
# level="DEBUG",
log_formatter=config.log_format,
log_file=log_file,
log_file_append=log_file_append,
)
return logger
def set_logger_from_dict(name: str, dict_values: dict) -> logging.Logger:
log_file = None
log_file_append = True
if dict_values.get('log_to_file'):
log_file = dict_values.get('log_file_path')
log_file_append = dict_values.get('log_file_append_mode')
if dict_values.get('debug'):
logger = setup_logger(
name=name,
level="DEBUG",
log_formatter=dict_values.get('log_format'),
log_file=log_file,
log_file_append=log_file_append,
)
else:
logger = setup_logger(
name=name,
# level="DEBUG",
log_formatter=dict_values.get('log_format'),
log_file=log_file,
log_file_append=log_file_append,
)
return logger
\ No newline at end of file
---
default:
debug: true
verbose: true
enabled_led: True
description: "Updated default app description"
boxes_used: "box_1, box_2"
box_1:
pin_button: 21
pin_led: 22
box_2:
pin_button: 23
pin_led: 23
###
### DEBUG
###
foo: "bar"
# boxes_used: "box_1, box_2, box_3"
# box_3:
# pin_button: 24
# pin_led: 25
#!/usr/bin/env python3
"""
utils.config.default_vals.py file for the c3 buttons.
Used for setting the default values from the 'utils.app_configuration.py' file.
"""
DEFAULT_VALUES = {
'DEFAULT_DESCRIPTION': 'UPDATE_THIS',
# 'DEFAULT_PATH_CONFIG_ENV': './.env',
'DEFAULT_PATH_CONFIG_ENV': '.env',
###
### config.yml
###
'DEFAULT_USING_PARSED_YAML': False,
'DEFAULT_PATH_CONFIG_YML': './utils/configuration/config.yml',
###
### DEBUG/VEBOSE
###
'DEFAULT_DEBUG': False,
'DEFAULT_VERBOSE': False,
###
### LOG
###
# "DEFAULT_LOG_TO_FILE": True,
"DEFAULT_LOG_TO_FILE": False,
"DEFAULT_LOG_FILE_APPEND_MODE": True,
"DEFAULT_LOG_FILE_PATH": "./logs/example.log",
"DEFAULT_LOG_FORMAT": '%(asctime)s - %(levelname)s - %(name)s - %(message)s',
###
### DB
###
"DEFAULT_DATABASE_TYPE":"local_file",
"DEFAULT_DATABASE_TABLE_NAME_BUTTON_BOX": "button_box",
"DEFAULT_DATABASE_TABLE_NAME_BUTTON_BOX_USE": "button_box_use",
"DEFAULT_DATABASE_LOCAL_DB_FILE_PATH":"Please set --> DATABASE_LOCAL_DB_FILE_PATH",
"DEFAULT_DATABASE_API_URL":"http://localhost:8000",
"DEFAULT_DATABASE_API_TOKEN":"Please set --> DATABASE_API_TOKEN",
###
### BOXES CONFIG
###
'DEFAULT_BOX_LED_ENABLED': False,
'DEFAULT_BOX_LED_AWAYS_ON': False,
'DEFAULT_BOX_LED_BLINK_TIMES': 5,
### DEFAULT_BOX_BUTTON_BOUNCE_TIME in microseconds
'DEFAULT_BOX_BUTTON_BOUNCE_TIME': "0.1",
###
### BOXES USED
###
"DEFAULT_BOXES_USED": "DEFAULT_BOX_1, DEFAULT_BOX_2, DEFAULT_BOX_3, DEFAULT_BOX_4, DEFAULT_BOX_5",
"DEFAULT_BOX_1": '{"pin_button": 14, "pin_led": 15}',
"DEFAULT_BOX_2": '{"pin_button": 18, "pin_led": 23}',
"DEFAULT_BOX_3": '{"pin_button": 24, "pin_led": 25}',
"DEFAULT_BOX_4": '{"pin_button": 8, "pin_led": 7}',
"DEFAULT_BOX_5": '{"pin_button": 12, "pin_led": 16}',
}
#!/usr/bin/env python3
"""
utils.configuration.from_file.py file for the c3 buttons.
This is the super class used from the other classes that fetches configurations from a file.
"""
import sys
from abc import ABC, abstractmethod
### LOCAL
try:
from .app_logger import set_logger_from_default
except ImportError as e:
print("Unable to import 'setup_logger' from 'utils.app_logger.py' file at 'utils.configuration.from_file.py' file" )
sys.exit(1)
class ConfigurationFromFile(ABC):
""" TODO
"""
def __init__(self, name: str, default_values: dict) -> None:
self.default_values = default_values
# self.logger = set_logger_from_default(name=__name__, values_default=self.default_values)
self.logger = set_logger_from_default(name=name, values_default=self.default_values)
self.path_config_file = None
self.dict_values = {}
@abstractmethod
def get_values(self):
pass
#!/usr/bin/env python3
"""
utils.configuration.from_file.py file for the c3 buttons.
"""
###
### DEPENDENCIES
###
import sys
import os
import json
try:
from dotenv import load_dotenv
except ImportError as e:
print("Unable to import 'dotenv'")
print("Please install it with 'pip(3) install dotenv --user'")
sys.exit(1)
### LOCAL
try:
from .from_file import ConfigurationFromFile
except ImportError as e:
print("Unable to import 'ConfigurationFromFile' from 'utils.configuration.from_file.py' file at 'utils.configuration.from_file_environment.py' file" )
sys.exit(1)
class ConfigurationFromDotEnv(ConfigurationFromFile):
""" TODO
"""
def __init__(self, default_values: dict) -> None:
"""
TODO: Documentation
"""
### Init super
super().__init__(name=__name__, default_values=default_values)
self.path_config_file = self.default_values.get('DEFAULT_PATH_CONFIG_ENV')
self.dict_values["boxes"] = {}
self.get_values()
def __get_env_bool__(self, name_config: str, name_config_default: str) -> bool:
"""
Method to fetch the environment variable from a boolean type
Returns
-------
bool:
Return a boolean value from fetching the setted environment variable
"""
self.logger.debug(f"__get_env_bool__ - {name_config}")
try:
fetched = os.getenv(name_config)
if fetched:
if fetched in ('YES', 'yes', 'true', 'True', 'TRUE'):
ret_bool = True
elif fetched in ('NO', 'no', 'false', 'False', 'FALSE'):
ret_bool = False
else:
print(f"\n\n Unable to fetch the parsed environment '{name_config}'. It is set but not 'true' or 'false' \n\n")
sys.exit(1)
else:
ret_bool = self.default_values.get(name_config_default)
except Exception as e:
print(f"\n\n Unable to fetch the parsed environment '{name_config}' \n| The exception is: '{e}' \n\n")
sys.exit(1)
return ret_bool
def __get_env__(self, name_config: str, name_config_default: str) -> (str | int):
"""
Method to fetch the environment variable from a string or int type.
Returns
-------
str | int:
Return a the value from fetching the setted environment variable.
"""
self.logger.debug(f"__get_env__ - {name_config}")
try:
fetched = os.getenv(name_config)
### If not setted, get it from config
if fetched is None:
fetched = self.default_values.get(name_config_default)
except Exception as e:
print(f"\n\n Unable to fetch the parsed environment '{name_config}' \n| The exception is: '{e}' \n\n")
sys.exit(1)
return fetched
def __get_env_dict__(self, name_config: str, name_config_default: str):
"""
Method to fetch the environment variable from a string or int type.
Returns
-------
str | int:
Return a the value from fetching the setted environment variable.
"""
self.logger.debug(f"__get_env_dict__ - {name_config}")
fetched = self.__get_env__(
name_config=name_config,
name_config_default=name_config_default
)
try:
ret_dict = json.loads(fetched)
except Exception as e:
print(f"\n\n Unable to parse the dictionary environment from: '{name_config}' \n| The exception is: '{e}' \n\n")
sys.exit(1)
self.dict_values["boxes"][name_config.lower()] = ret_dict
def get_values(self) -> None:
"""
Used for fetching the parsed values from the system.
If the value is not parsed, it will use the default values settend at: utils.configuration.default_vals.py
"""
self.logger.debug("get_values")
if not load_dotenv(dotenv_path=self.path_config_file):
print("\n\n ERROR!: \ņ\n Unable to load the '.env' file at the path: '{self.path_config_file}'. Change it at the '.configuration.default_vals.py' file")
sys.exit()
self.dict_values["description"] = self.__get_env__(
name_config="DESCRIPTION",
name_config_default="DEFAULT_DESCRIPTION"
)
###
### DEBUG/VEBOSE
###
self.dict_values["debug"] = self.__get_env_bool__(
name_config="DEBUG",
name_config_default="DEFAULT_DEBUG"
)
self.dict_values["verbose"] = self.__get_env_bool__(
name_config="VERBOSE",
name_config_default="DEFAULT_VERBOSE"
)
###
### LOG
###
self.dict_values["log_file_append_mode"] = self.__get_env_bool__(
name_config="LOG_FILE_APPEND_MODE",
name_config_default="DEFAULT_LOG_FILE_APPEND_MODE"
)
self.dict_values["log_to_file"] = self.__get_env_bool__(
name_config="LOG_TO_FILE",
name_config_default="DEFAULT_LOG_TO_FILE"
)
self.dict_values["log_file_path"] = self.__get_env__(
name_config="LOG_FILE_PATH",
name_config_default="DEFAULT_LOG_FILE_PATH"
)
self.dict_values["log_format"] = self.__get_env__(
name_config="LOG_FORMAT",
name_config_default="DEFAULT_LOG_FORMAT"
)
###
### DB
###
database_type = self.__get_env__(
name_config="DATABASE_TYPE",
name_config_default="DEFAULT_DATABASE_TYPE"
)
if database_type == "local_file":
self.dict_values["database_table_name_button_box"] = self.__get_env__(
name_config="DATABASE_TABLE_NAME_BUTTON_BOX",
name_config_default="DEFAULT_DATABASE_TABLE_NAME_BUTTON_BOX"
)
self.dict_values["database_table_name_button_box_use"] = self.__get_env__(
name_config="DATABASE_TABLE_NAME_BUTTON_BOX_USE",
name_config_default="DEFAULT_DATABASE_TABLE_NAME_BUTTON_BOX_USE"
)
self.dict_values["database_local_db_file_path"] = self.__get_env__(
name_config="DATABASE_LOCAL_DB_FILE_PATH",
name_config_default="DEFAULT_DATABASE_LOCAL_DB_FILE_PATH"
)
elif database_type == "api":
self.dict_values["database_api_url"] = self.__get_env__(
name_config="DATABASE_API_URL",
name_config_default="DEFAULT_DATABASE_API_URL"
)
self.dict_values["database_api_token"] = self.__get_env__(
name_config="DATABASE_API_TOKEN",
name_config_default="DEFAULT_DATABASE_API_TOKEN"
)
else:
print(f"\n\n Unable to set the database_type from '{database_type}'\n Accepted values are: 'local_file' or 'api' \n\n")
sys.exit(1)
self.dict_values["database_type"] = database_type
###
### BOXES CONFIG
###
self.dict_values["box_button_bounce_time"] = self.__get_env__(
name_config="BOX_BUTTON_BOUNCE_TIME",
name_config_default="DEFAULT_BOX_BUTTON_BOUNCE_TIME"
)
self.dict_values["box_led_enabled"] = self.__get_env_bool__(
name_config="BOX_LED_ENABLED",
name_config_default="DEFAULT_BOX_LED_ENABLED"
)
self.dict_values["box_led_always_on"] = self.__get_env_bool__(
name_config="BOX_LED_AWAYS_ON",
name_config_default="DEFAULT_BOX_LED_AWAYS_ON"
)
self.dict_values["box_led_blink_times"] = self.__get_env__(
name_config="BOX_LED_BLINK_TIMES",
name_config_default="DEFAULT_BOX_LED_BLINK_TIMES"
)
###
### BOXES USED
###
# Set boxes_used as a list
self.dict_values["boxes_used"] = self.__get_env__(
name_config="BOXES_USED",
name_config_default="DEFAULT_BOXES_USED"
).replace(' ','').replace("'", "").replace('"', '').split(',')
# import pdb; pdb.set_trace()
# self.dict_values["boxes_used"]
# os.getenv("BOXES_USED")
# load_dotenv(dotenv_path=self.path_config_file)
# # "'BUTTON_BOX_1, BUTTON_BOX_2'"
# Adds dict_values with 'boxes' infos from 'boxes_used' list
for box_name in self.dict_values.get("boxes_used"):
self.app_description = self.__get_env_dict__(
name_config=box_name,
name_config_default="DEFAULT_" + box_name
)
# Fix lower boxes names from boxes_used list
for index, item in enumerate(self.dict_values.get("boxes_used")):
self.dict_values.get("boxes_used")[index] = item.lower()
#!/usr/bin/env python3
"""
utils.configuration.from_file_yaml.py file for the c3 buttons.
"""
###
### DEPENDENCIES
###
import sys
import json
try:
import pyaml_env
except ImportError as e:
print("Unable to import 'pyaml_env'")
print("Please install it with 'pip(3) install pyaml_env --user'")
sys.exit(1)
### LOCAL
try:
from .from_file import ConfigurationFromFile
except ImportError as e:
print("Unable to import 'ConfigurationFromFile' from 'utils.configuration.from_file.py' file at the 'utils.configuration.from_file_yaml.py' file" )
sys.exit(1)
class ConfigurationFromYamlFile(ConfigurationFromFile):
def __init__(self, default_values: dict) -> None:
### Init super
super().__init__(name=__name__, default_values=default_values)
self.path_config_file = self.default_values.get('DEFAULT_PATH_CONFIG_YML')
self.dict_values = self.get_values()
def __get_env_dict__(self, parsed: dict) -> dict:
self.logger.debug(f"__get_env_dict__ - {parsed}")
parsed['boxes_used'] = parsed.get("boxes_used").replace(' ',''). split(',')
parsed['boxes'] = {}
for box_name in parsed.get("boxes_used"):
box_info = parsed.get(box_name)
if box_info is None:
print(f"\n\n Error getting the configuration from the box '{box_name}' from the 'config.yml' file. \n If you want to use this box, configure it also.\n\n")
sys.exit(1)
if isinstance(box_info, dict):
ret_dict = box_info
else:
try:
ret_dict = json.loads(box_info)
except Exception as e:
print(f"\n\n Unable to parse the dictionary from the '{box_name}' with value: '{box_info}' \n| The exception is: '{e}' \n\n")
sys.exit(1)
parsed["boxes"][box_name.lower()] = ret_dict
parsed.pop(box_name)
return parsed
def get_values(self) -> dict:
self.logger.debug(f"get_values")
try:
parsed_all = pyaml_env.parse_config(self.path_config_file)
except Exception as e:
print(f"\n\n Error reading the 'yaml' configurtion file. \n Exception: {e}")
sys.exit(1)
default_parsed = parsed_all.get('default')
if default_parsed is None:
print(f"\n\n Error getting the configuration from the 'config.yml' file. \n Exception: {e}")
sys.exit(1)
return self.__get_env_dict__(parsed=default_parsed)
#!/usr/bin/env python3
"""
utils.database.app_database.py file for the c3 printing system.
"""
import sys
import logging
### LOCAL
try:
from utils.configuration.app_logger import set_logger_from_config
except ImportError as e:
print("Unable to import 'setup_logger' from 'utils.configuration.app_logger.py' file at 'utils.database.app_database.py' file")
sys.exit(1)
try:
from .from_local_file import DataBaseLocalSqLite
except ImportError as e:
print("Unable to import 'DataBaseUsed' from 'utils.database.from_local_file.py' file at 'utils.database.app_database.py' file")
sys.exit(1)
try:
from .from_api import DataBaseApi
except ImportError as e:
print("Unable to import 'DataBaseApi' from 'utils.database.from_api.py' file at 'utils.database.app_database.py' file")
sys.exit(1)
class AppDatabase():
"""
Class for setting the configuration of the app.
"""
def __init__(self, configuration) -> None:
"""
Function for initialiazing the AppConfiguration class.
Sets the class variables.
Returns
-------
None
"""
self.configuration = configuration
self.logger = set_logger_from_config(name=__name__, config=self.configuration)
self.set_database()
def set_database(self) -> None:
self.logger.debug("set_database")
if self.configuration.database_type == "local_file":
self.logger.info("---> Using local SQLite3 file as Database")
self.db = DataBaseLocalSqLite(configuration=self.configuration)
elif self.configuration.database_type == "api":
self.logger.info("---> Using API calls as Database")
self.db = DataBaseApi(configuration=self.configuration)
else:
print(f"\n\n ERROR! ---> Unable to set database from the parsed 'database_type' with value: {self.configuration.database_type}")
sys.exit(1)
#!/usr/bin/env python3
"""
utils.database.from_api.py file for the c3 buttons.
"""
###
### DEPENDENCIES
###
import sys
import requests
### LOCAL
try:
from .from_database import DataBaseUsed
except ImportError as e:
print("Unable to import 'DataBaseUsed' from 'utils.database.from_database.py' file at 'utils.database.from_api.py' file" )
sys.exit(1)
class DataBaseApi(DataBaseUsed):
"""
TODO: Documentation
"""
def __init__(self, configuration: dict) -> None:
"""
TODO: Documentation
"""
### Init super
super().__init__(name=__name__, configuration=configuration)
self.__check_token__()
def __check_token__(self) -> None:
"""
TODO: Documentation
"""
self.logger.debug("__check_token__")
response = self.get(endpoint="auth/test/closed")
if response is None:
self.logger.error("__check_token__ - Token is NOT working...")
print("\nERROR ---> Initial Token is not working.\n You should Generate a new one and set in the configuration...\n\n")
sys.exit()
else:
if response.get('error'):
print("\nERROR ---> Some error happend trying to get the initial token.\n ")
sys.exit()
else:
self.logger.info("__check_token__ - Token is working...")
def __get_token_headers__(self) -> dict:
return {
"Content-Type": "application/json; charset=utf-8",
"Authorization": f"Bearer {self.configuration.database_api_token}"
}
def get(self, endpoint, params=None) -> (dict | None):
self.logger.debug(f"get - endpoint: '{endpoint}' | params: '{params}'")
url = f"{self.configuration.database_api_url}/{endpoint}"
try:
response = requests.get(
url,
headers=self.__get_token_headers__(),
params=params
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
self.logger.error(f"GET request failed with exception: {e}")
return None
def post(self, endpoint, json=None, data=None) -> (dict | None):
self.logger.debug(f"post - endpoint: '{endpoint}' | json: '{json}' | data: '{data}'")
url = f"{self.configuration.database_api_url}/{endpoint}"
try:
response = requests.post(
url,
json=json,
data=data,
headers=self.__get_token_headers__()
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
self.logger.error(f"POST request failed with exception: {e}")
return None
def add_button_box_usage(self, box_name: str):
self.logger.debug(f"add_button_box_usage - box_name: '{box_name}'")
post_data = {
"name": box_name
}
resp = self.post(endpoint="api/box/use", json=post_data)
if resp.get('error'):
self.logger.error(f"Error adding button usage with response: {resp}")
#!/usr/bin/env python3
"""
utils.configuration.from_file.py file for the c3 buttons.
This is the super class used from the other classes that fetches configurations from a file.
"""
import sys
from abc import ABC, abstractmethod
try:
from utils.configuration.app_logger import set_logger_from_config
except ImportError as e:
print("Unable to import 'set_logger_from_dict' from 'utils.configuration.app_logger.py' file at 'utils.database.from_api.py' file" )
sys.exit(1)
class DataBaseUsed(ABC):
""" TODO
"""
def __init__(self, name: str, configuration: dict) -> None:
self.configuration = configuration
# self.logger = set_logger_from_config(name=__name__, config=self.configuration)
self.logger = set_logger_from_config(name=name, config=self.configuration)
self.debug = configuration.debug
self.conn = None
self.cursor = None
@abstractmethod
def add_button_box_usage(self, box_name: str):
pass
#!/usr/bin/env python3
"""
utils.database.from_local_file.py file for the c3 buttons.
"""
###
### DEPENDENCIES
###
import sys
import sqlite3
from datetime import datetime
### LOCAL
try:
from .from_database import DataBaseUsed
except ImportError as e:
print("Unable to import 'DataBaseUsed' from 'utils.database.from_database.py' file at 'utils.database.from_local_file.py' file" )
sys.exit(1)
class DataBaseLocalSqLite(DataBaseUsed):
""" TODO
"""
def __init__(self, configuration: dict) -> None:
### Init super
super().__init__(name=__name__, configuration=configuration)
self.file_path = self.configuration.database_local_db_file_path
###
### CONNECTION
###
def conn_create(self):
self.conn = sqlite3.connect(self.file_path)
self.cursor = self.conn.cursor()
def conn_close(self):
if self.conn:
self.cursor.close()
self.conn.close()
self.cursor = None
self.conn = None
###
### QUERY
###
def query_fetch_one(self, query: str) -> None:
# self.logger.debug(f"query_fetch_one - query: {query}")
self.logger.debug("query_fetch_one")
if self.conn is None:
self.conn_create()
try:
resp = self.cursor.execute(query)
return resp.fetchone()[0]
except Exception as e:
self.logger.error(f"ERROR executing query: {query} | with execption: {e}")
def query_fetch_all(self, query: str) -> None:
# self.logger.debug(f"query_fetch_all - query: {query}")
self.logger.debug("query_fetch_all")
if self.conn is None:
self.conn_create()
try:
self.cursor.execute(query)
resp = self.cursor.execute(query)
return resp.fetchall()
except Exception as e:
self.logger.error(f"ERROR executing query: {query} | with execption: {e}")
finally:
self.conn_close()
def query_exec(self, query: str) -> None:
# self.logger.debug(f"query_exec - query: {query}")
self.logger.debug("query_exec")
if self.conn is None:
self.conn_create()
try:
self.cursor.execute(query)
except Exception as e:
self.logger.error(f"ERROR executing query: {query} | with execption: {e}")
finally:
if self.conn:
self.conn.commit()
###
### TABLES
###
def table_select_all(self, table_name: str) -> (list | None):
self.logger.debug(f"table_select_all - table_name: {table_name}")
if self.conn is None:
self.conn_create()
query = "SELECT * FROM " + table_name + ";"
return self.query_fetch_all(query=query)
def table_get_button_box_id(self, box_name: str):
self.logger.debug(f"table_get_button_box_id - box_name '{box_name}'")
table_name = self.configuration.database_table_name_button_box
query = """
SELECT id FROM {table_name} WHERE {field_name}="{field_value}";
""".format(
table_name=table_name,
field_name="name",
field_value=box_name
)
# box_id = self.query_fetch_one(query=query)
# self.conn_close()
# return box_id
##################
##################
##################
return self.query_fetch_one(query=query)
def add_button_box_usage(self, box_name: str):
self.logger.debug(f"table_add_button_box_usage - box_name '{box_name}'")
if self.conn is None:
self.conn_create()
button_box_id = self.table_get_button_box_id(box_name=box_name)
values = (str(datetime.now()), button_box_id)
query = """
INSERT INTO "{table_name}" {columns_list} VALUES {values}
""".format(
table_name=self.configuration.database_table_name_button_box_use,
columns_list=("date", "box_used_id"),
values=values
)
self.query_exec(query)
self.conn_close()
[pycodestyle]
count = False
; E266 too many leading '#' for block comment
; E401 multiple imports on one line
ignore = E266, E401
max-line-length = 160
; max-line-length = 81
statistics = True
\ No newline at end of file