<repo_name>AlexanderMg12/expenses-table<file_sep>/config.py
HOST = "test"
USER = "test"
PASSWORD = "<PASSWORD>"
DATABASE = "test"
TABLE_NAME = "expenses"
ERROR_PAGE = "Error_page.html"
REDIRECT_PAGE = "index"
<file_sep>/expenses.py
from typing import Union

from flask_bootstrap import Bootstrap
from flask import Flask, request, redirect, url_for, render_template
from mysql.connector import Error
from werkzeug.wrappers.response import Response

from config import ERROR_PAGE, REDIRECT_PAGE
from sql_handler import (
    get_all_data,
    get_data_to_id,
    add_line,
    delete_line_to_id,
    delete_all_lines,
    update_line,
)
from logger import logger_expenses
from utils import get_dollar_value, get_line_to_id


app = Flask(__name__)
Bootstrap(app)


@app.route("/", methods=["GET"])
def index() -> str:
    """Основной route приложения.

    Returns:
        str: html страницы.
    """
    try:
        all_data = get_all_data()
        return render_template("index.html", data=all_data)
    except Error as e:
        logger_expenses.error(f"Ошибка: {e}", exc_info=True)
        return render_template(ERROR_PAGE)


@app.route("/add_data", methods=["POST"])
def add_data() -> Union[Response, str]:
    """Добавление строки с данными в таблицу.

    Returns:
        Union[Response, str]: редирект на основную html или html-страница.
    """
    form = request.form
    rubles = float(form["rub"])
    waste_or_income = form["w/i"]
    description = form["desc"]
    dollars = round(rubles / get_dollar_value(), 2)

    try:
        add_line(rubles, dollars, waste_or_income, description)
        logger_expenses.debug(
            "Добавлена строка с данными: "
            f"rubles = {rubles}, dollars = {dollars}, "
            f"w/i = {waste_or_income}, description = {description}"
        )
        return redirect(url_for(REDIRECT_PAGE))
    except Error as e:
        logger_expenses.error(f"Ошибка: {e}", exc_info=True)
        return render_template(ERROR_PAGE)


@app.route("/delete_data", methods=["POST"])
def delete_data() -> Union[Response, str]:
    """Удаление строки с данными в таблице.

    Returns:
        Union[Response, str]: редирект на основную html или html-страница.
    """
    line_id = int(request.form["id"])
    delete_line = get_data_to_id(line_id)[0]
    logger_expenses.debug(
        f"Запрос на удаление строки с id {line_id}, данные строки: "
        f"rubles = {delete_line[1]}, dollars = {delete_line[2]}, date = {delete_line[3]}, "
        f"w/i = {delete_line[4]}, description = {delete_line[5]}"
    )

    try:
        delete_line_to_id(line_id)
        logger_expenses.debug(f"Удалена строка с id {line_id}")
        return redirect(url_for(REDIRECT_PAGE))
    except Error as e:
        logger_expenses.error(f"Ошибка: {e}", exc_info=True)
        return render_template(ERROR_PAGE)


@app.route("/delete_all_data", methods=["POST"])
def delete_all_data() -> Union[Response, str]:
    """Удаление всех строк в таблице.

    Returns:
        Union[Response, str]: редирект на основную html или html-страница.
    """
    try:
        delete_all_lines()
        logger_expenses.debug("Все данные удалены")
        return redirect(url_for(REDIRECT_PAGE))
    except Error as e:
        logger_expenses.error(f"Ошибка: {e}", exc_info=True)
        return render_template(ERROR_PAGE)


@app.route("/update_data", methods=["POST"])
def update_data() -> Union[Response, str]:
    """Route для изменения строки с данными в таблицу.

    Returns:
        Union[Response, str]: редирект на основную html или html-страница.
    """
    line_id = int(request.form["id"])
    line_to_id = get_line_to_id(line_id)
    logger_expenses.debug(
        f"Запрос на изменение данных в строке с id {line_id}: "
        f"rubles = {line_to_id[1]}, dollars = {line_to_id[2]}, date = {line_to_id[3]}, "
        f"w/i = {line_to_id[4]}, description = {line_to_id[5]}"
    )

    form = request.form
    rubles = float(form["rub"]) if form["rub"] else line_to_id[1]
    waste_or_income = form["w/i"] if form["w/i"] else line_to_id[4]
    description = form["desc"] if form["desc"] else line_to_id[5]
    dollars = round(rubles / get_dollar_value(), 2)

    try:
        update_line(rubles, waste_or_income, description, dollars, line_id)
        line_to_id = get_line_to_id(line_id)
        logger_expenses.debug(
            "Данные изменены на: "
            f"rubles = {line_to_id[1]}, dollars = {line_to_id[2]}, date = {line_to_id[3]}, "
            f"w/i = {line_to_id[4]}, description = {line_to_id[5]}"
        )
        return redirect(url_for(REDIRECT_PAGE))
    except Error as e:
        logger_expenses.error(f"Ошибка: {e}", exc_info=True)
        return render_template(ERROR_PAGE)


if __name__ == "__main__":
    app.run()
<file_sep>/requirements.txt
Flask==1.1.2
redis==5.0.7
requests==2.22.0
flask_bootstrap==3.3.7.1
mysql_connector_repackaged==0.3.1
<file_sep>/logger.py
import logging


logging.basicConfig(level=logging.DEBUG)
logger_expenses = logging.getLogger("expenses")
my_handler = logging.FileHandler("log_table.log")
my_format = logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s")
my_handler.setFormatter(my_format)
logger_expenses.addHandler(my_handler)
logger_expenses.setLevel(logging.DEBUG)
<file_sep>/redis_handler.py
import redis
from typing import Union

redis_conn = redis.Redis()


def recording_dollar_value(dollar_value: str) -> None:
    """Запись значения доллара

    Args:
        dollar_value: значение доллара
    """
    redis_conn.setex("dollar", 3600, dollar_value)


def check_dollar_value() -> Union[bytes, None]:
    """Проверка наличия значения доллара в кэш

    Return:
        Union[bytes, None]: значение доллара или ничего
    """
    return redis_conn.get("dollar")
<file_sep>/create_table.py
import mysql.connector
from mysql.connector import Error


conn = mysql.connector.connect(
    host="localhost", user="alexandr", password="1", database="mysqltest"
)

cursor = conn.cursor()
create_table_data = """
CREATE TABLE IF NOT EXISTS expenses (
id INT AUTO_INCREMENT,
rubles FLOAT NOT NULL,
dollars FLOAT NOT NULL,
date_time DATETIME NOT NULL,
waste_or_income TEXT NOT NULL,
description TEXT NOT NULL,
PRIMARY KEY (id)
)
"""
try:
    cursor.execute(create_table_data)
    conn.commit()
    print("Таблица создана")
except Error as e:
    print(e)
<file_sep>/sql_handler.py
import mysql.connector
from typing import List

from config import HOST, USER, PASSWORD, DATABASE, TABLE_NAME

mysql_conn = mysql.connector.connect(
    host=HOST, user=USER, password=<PASSWORD>, database=DATABASE
)
mysql_cursor = mysql_conn.cursor()


def get_all_data() -> List[tuple]:
    """Получение всех данных из таблицы.

    Return:
         List[tuple]: Все данные таблицы.
    """
    query = f"SELECT * FROM {TABLE_NAME}"
    mysql_cursor.execute(query)
    return mysql_cursor.fetchall()


def get_data_to_id(line_id: int) -> List[tuple]:
    """Получение значений из таблицы по заданному id.

    Args:
        line_id: id строки в таблице.

    Return:
        List[tuple]: Данные строки из таблицы по заданному id.
    """
    query = f"SELECT * FROM {TABLE_NAME} WHERE id = {line_id}"
    mysql_cursor.execute(query)
    return mysql_cursor.fetchall()


def add_line(
    rubles: float, dollars: float, waste_or_income: str, description: str
) -> None:
    """Добавление новой строки в таблицу.

    Args:
        rubles: значение суммы в рублях.
        dollars: значение суммы в долларах.
        waste_or_income: вид транзакции (расход/доход).
        description: описание транзакции.
    """
    query = """INSERT INTO expenses
    ( rubles, dollars, date_time, waste_or_income, description )
    VALUES ( %s, %s, NOW(), %s, %s)"""
    mysql_cursor.execute(query, (rubles, dollars, waste_or_income, description))
    mysql_conn.commit()


def delete_line_to_id(line_id: int) -> None:
    """Удаление строки в таблице по заданному id.

    Args:
        line_id: id строки в таблице.
    """
    query = f"DELETE FROM {TABLE_NAME} WHERE id = {line_id}"
    mysql_cursor.execute(query)
    mysql_conn.commit()


def delete_all_lines() -> None:
    """Удаление всех данных таблицы."""
    query = f"DELETE FROM {TABLE_NAME}"
    mysql_cursor.execute(query)
    mysql_conn.commit()


def update_line(
    rubles: float, waste_or_income: str, description: str, dollars: float, line_id: int
) -> None:
    """Обновление данных строки в таблице.

    Args:
        rubles: рубли.
        waste_or_income: расход или доход.
        description: описание.
        dollars: доллары.
        line_id: id строки в таблице.

    """
    query = """UPDATE expenses SET
        rubles = %s,
        waste_or_income = %s,
        description = %s,
        dollars = %s,
        date_time = NOW()
        WHERE id = %s"""
    mysql_cursor.execute(
        query, (rubles, waste_or_income, description, dollars, line_id)
    )
    mysql_conn.commit()
<file_sep>/utils.py
import re
from typing import Union

import requests
from flask import render_template
from mysql.connector import Error
from werkzeug.wrappers.response import Response

from redis_handler import recording_dollar_value, check_dollar_value
from sql_handler import get_data_to_id
from logger import logger_expenses
from config import ERROR_PAGE


def get_dollar_value() -> float:
    """Получаем значения доллара из redis, если значения нет, парсим сайт,
    для получения значения и записываем его в redis.

    Return:
        float: значение доллара.
    """
    dollar_value_cache = check_dollar_value()
    if dollar_value_cache is None:
        dollar_html = requests.get("https://www.banki.ru/products/currency/usd/")
        dollar_value = re.search(
            r'<div class="currency-table__large-text">(\d+.\d+)', dollar_html.text
        ).group(1)
        dollar_value = dollar_value.replace(",", ".")
        recording_dollar_value(dollar_value)
    else:
        dollar_value = dollar_value_cache.decode("utf-8", "replace")

    return float(dollar_value)


def get_line_to_id(line_id: int) -> Union[tuple, Response]:
    """Получение кортежа с данными строки из таблицы базы данных.

    Args:
        line_id: id строки в таблице.

    Return:
        Union[tuple, Response]: кортеж с данными строки из таблицы или страницу с ошибкой
    """
    try:
        return get_data_to_id(line_id)[0]
    except Error as err:
        logger_expenses.error(f"Ошибка: {err}", exc_info=True)
        return render_template(ERROR_PAGE)
