Основной код проекта - nemopss/mpt-kpi GitHub Wiki

1.4.17 routes.py

В данном файле routes.py определяются маршруты (routes) приложения Flask, которые используются для обработки HTTP-запросов. Он обеспечивает функциональность работы с пользователями и позициями, реализуя обработку запросов на получение и создание ресурсов, а также контроль доступа с помощью проверки административных прав текущего пользователя.

from flask import Blueprint, jsonify, request
from flask_login import current_user, login_required

from extensions import db
from models import Position, User

bp = Blueprint("main", __name__)

@bp.route("/users", methods=["GET", "POST"])
@login_required
def users():
    if request.method == "GET":
        users = User.query.all()
        return jsonify([{"id": u.id, "email": u.email} for u in users])
    elif request.method == "POST":
        if not current_user.is_admin:
            return jsonify({"error": "Access denied"}), 403
        data = request.json
        user = User(
            first_name=data["first_name"],
            last_name=data["last_name"],
            email=data["email"],
        )
        db.session.add(user)
        db.session.commit()
        return jsonify({"message": "User created"}), 201

@bp.route("/positions", methods=["GET", "POST"])
@login_required
def positions():
    if not current_user.is_admin:
        return jsonify({"error": "Access denied"}), 403

    if request.method == "GET":
        positions = Position.query.all()
        return jsonify([{"id": p.id, "name": p.name} for p in positions])
    elif request.method == "POST":
        data = request.json
        position = Position(name=data["name"], score_threshold=data["score_threshold"])
        db.session.add(position)
        db.session.commit()
        return jsonify({"message": "Position created"}), 201

Импортируемые модули

from flask import Blueprint, jsonify, request
from flask_login import current_user, login_required

from extensions import db
from models import Position, User

• Blueprint: Используется для создания модульной структуры приложения Flask.

• jsonify: Функция для преобразования Python-объектов в JSON-формат.

• request: Объект, который содержит данные текущего HTTP-запроса.

• current_user: Представляет текущего аутентифицированного пользователя.

• login_required: Декоратор, который гарантирует, что маршруты доступны только для аутентифицированных пользователей.

• db: Расширенный объект базы данных для работы с SQLAlchemy.

• User и Position: Модели, представляющие пользователей и позиции в базе данных соответственно.


Инициализация Blueprint

bp = Blueprint("main", __name__)

• Создается объект Blueprint с именем "main", который будет использоваться для группировки связанных маршрутов.


Маршрут /users

Метод GET

@bp.route("/users", methods=["GET", "POST"])
@login_required
def users():
    if request.method == "GET":
        users = User.query.all()
        return jsonify([{"id": u.id, "email": u.email} for u in users])

• Этот маршрут обрабатывает запросы для получения списка пользователей. При выполнении GET-запроса возвращается список всех пользователей с их идентификаторами и электронными адресами в формате JSON.


Метод POST

    elif request.method == "POST":
        if not current_user.is_admin:
            return jsonify({"error": "Access denied"}), 403
        data = request.json
        user = User(
            first_name=data["first_name"],
            last_name=data["last_name"],
            email=data["email"],
        )
        db.session.add(user)
        db.session.commit()
        return jsonify({"message": "User created"}), 201

• Этот маршрут позволяет создавать новых пользователей. Доступ к этому маршруту имеют только администраторы. При выполнении POST-запроса требуется передать данные нового пользователя в формате JSON. Если текущий пользователь не является администратором, возвращается ошибка доступа.


Маршрут /positions

Метод GET

@bp.route("/positions", methods=["GET", "POST"])
@login_required
def positions():
    if not current_user.is_admin:
        return jsonify({"error": "Access denied"}), 403

    if request.method == "GET":
        positions = Position.query.all()
        return jsonify([{"id": p.id, "name": p.name} for p in positions])

• Этот маршрут обрабатывает запросы для получения списка позиций. Доступен только для администраторов. При выполнении GET-запроса возвращает список всех позиций с их идентификаторами и названиями в формате JSON.


Метод POST

elif request.method == "POST":
        data = request.json
        position = Position(name=data["name"], score_threshold=data["score_threshold"])
        db.session.add(position)
        db.session.commit()
        return jsonify({"message": "Position created"}), 201

• Этот маршрут позволяет создавать новые позиции. При выполнении POST-запроса требуется передать данные о новой позиции в формате JSON. Позволяет добавить название позиции и пороговое значение баллов.

1.4.18 main.py

В файле main.py осуществляется настройка и инициализация основного приложения Flask, который представляет собой ядро приложения и отвечает за его функционирование и доступность. Здесь реализована конфигурация для работы с RESTful API, инициализация необходимых расширений, регистрация маршрутов и запуск сервера.

from flask import Flask, render_template
from flask_cors import CORS
from flask_restful import Api
from os import environ

from config import Config
from extensions import db, migrate, login_manager, mail
from resources.basic_resource import BasicResource
from resources.pdf_resource import PDFResource
from resources.employees_resource import EmployeesResource
from resources.criteries_resource import CriteriesResource
from resources.certificates_resource import CertificatesResource
from auth import auth_bp
from routes import bp as main_bp

PORT = environ.get('BACKEND_PORT', 8000)
DEBUG = environ.get('DEBUG', True)

app = Flask(__name__)
app.config.from_object(Config)

cors = CORS(app)
app.config['CORS_HEADERS'] = 'Content-Type'
app.config['MAX_CONTENT_LENGTH'] = 5 * 1024 * 1024  # 5 MB limit for uploads

api = Api(app)
api.add_resource(BasicResource, '/api', '/api/<int:source_id>')
api.add_resource(PDFResource, '/api/media/', '/api/media/<string:filename>')
api.add_resource(EmployeesResource, '/api/employees/', '/api/employees/')
api.add_resource(CriteriesResource, '/api/criteries/', '/api/criteries/')
api.add_resource(CertificatesResource, '/api/certificates/', '/api/certificates/')

# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
login_manager.init_app(app)
mail.init_app(app)

# Set up login behavior
@login_manager.user_loader
def load_user(user_id):
    from backend.models import User
    return User.query.get(int(user_id))

login_manager.login_view = "auth.login"
login_manager.login_message = "Please log in to access this page."

# Register blueprints
app.register_blueprint(main_bp)
app.register_blueprint(auth_bp)

@app.route('/', methods=['GET'])
def index():
    return render_template('index.html')

PORT = environ.get('BACKEND_PORT', 8000)
DEBUG = environ.get('DEBUG', True)
app.run(extra_files=[], debug=DEBUG, host='0.0.0.0', port=PORT)

Импортируемые модули

from flask import Flask, render_template
from flask_cors import CORS
from flask_restful import Api
from os import environ

from config import Config
from extensions import db, migrate, login_manager, mail
from resources.basic_resource import BasicResource
from resources.pdf_resource import PDFResource
from resources.employees_resource import EmployeesResource
from resources.criteries_resource import CriteriesResource
from resources.certificates_resource import CertificatesResource
from auth import auth_bp
from routes import bp as main_bp

• Flask: Основной класс приложения Flask.

• render_template: Функция для рендеринга HTML-шаблонов.

• CORS: Расширение для настройки кросс-доменных запросов.

• Api: Класс для создания RESTful API с использованием Flask-RESTful.

• environ: Модуль для работы с переменными окружения.


Конфигурация приложения

PORT = environ.get('BACKEND_PORT', 8000)
DEBUG = environ.get('DEBUG', True)

app = Flask(__name__)
app.config.from_object(Config)

• Переменные окружения BACKEND_PORT и DEBUG используются для настройки порта, на котором будет запущен сервер, и режима отладки.

• Конфигурация приложения загружается из класса Config.


Настройка CORS

cors = CORS(app)
app.config['CORS_HEADERS'] = 'Content-Type'
app.config['MAX_CONTENT_LENGTH'] = 5 * 1024 * 1024  # 5 MB limit for uploads

• CORS: Настраивает поддержку кросс-доменных запросов для приложения.

• Устанавливается заголовок Content-Type и ограничение на размер загружаемых данных (5 MB).


Регистрация RESTful ресурсов

api = Api(app)
api.add_resource(BasicResource, '/api', '/api/<int:source_id>')
api.add_resource(PDFResource, '/api/media/', '/api/media/<string:filename>')
api.add_resource(EmployeesResource, '/api/employees/', '/api/employees/')
api.add_resource(CriteriesResource, '/api/criteries/', '/api/criteries/')
api.add_resource(CertificatesResource, '/api/certificates/', '/api/certificates/')

• Создается объект Api, который регистрирует ресурсы с соответствующими маршрутами для обработки запросов.


Инициализация расширений

# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
login_manager.init_app(app)
mail.init_app(app)

• Расширения (такие как db, migrate, login_manager, и mail) инициализируются с использованием экземпляра приложения app.


Настройка пользовательской аутентификации

@login_manager.user_loader
def load_user(user_id):
    from backend.models import User
    return User.query.get(int(user_id))

login_manager.login_view = "auth.login"
login_manager.login_message = "Please log in to access this page."

• Функция load_user загружает пользователя из базы данных по его идентификатору.

• Устанавливаются параметры для управления поведением аутентификации.


Регистрация Blueprints

# Register blueprints
app.register_blueprint(main_bp)
app.register_blueprint(auth_bp)

• Регистрируются ранее созданные Blueprint для основной логики приложения (main_bp) и аутентификации (auth_bp).


Главная Страница

@app.route('/', methods=['GET'])
def index():
    return render_template('index.html')

• Определяется маршрут для главной страницы, которая рендерит шаблон index.html.


Запуск приложения

PORT = environ.get('BACKEND_PORT', 8000)
DEBUG = environ.get('DEBUG', True)
app.run(extra_files=[], debug=DEBUG, host='0.0.0.0', port=PORT)

• Приложение запускается на заданном порту и в заданном режиме отладки. Установлен хост 0.0.0.0, что позволяет приложению быть доступным извне.

1.4.19 auth.py

Файл auth.py реализует функционал аутентификации пользователей в приложении Flask и содержит маршруты, необходимые для входа в систему, выхода из нее и сброса пароля через отправку электронных писем.

from flask import Blueprint, jsonify, request, url_for
from flask_login import login_user, logout_user
from flask_mail import Message

from extensions import db, mail
from models import User

auth_bp = Blueprint("auth", __name__)

@auth_bp.route("/login", methods=["POST"])
def login():
    data = request.json
    user = User.query.filter_by(email=data["email"]).first()
    if user and user.check_password(data["password"]):
        login_user(user)
        return jsonify({"message": "Login successful"}), 200
    return jsonify({"error": "Invalid credentials"}), 401

@auth_bp.route("/logout", methods=["POST"])
def logout():
    logout_user()
    return jsonify({"message": "Logout successful"}), 200

@auth_bp.route("/reset_password", methods=["POST"])
def reset_password():
    data = request.json
    user = User.query.filter_by(email=data["email"]).first()
    if not user:
        return jsonify({"error": "User not found"}), 404

    token = user.id  # Using ID as an example token
    reset_url = url_for("auth.complete_reset", token=token, _external=True)

    msg = Message("Password Reset Request", recipients=[user.email])
    msg.body = f"To reset your password, visit the following link: {reset_url}"
    mail.send(msg)

    return jsonify({"message": "Reset link sent"}), 200

@auth_bp.route("/reset_password/<token>", methods=["POST"])
def complete_reset(token):
    user = User.query.get(int(token))
    if not user:
        return jsonify({"error": "Invalid token"}), 400

    data = request.json
    user.set_password(data["password"])
    db.session.commit()
    return jsonify({"message": "Password reset successful"}), 200

Импортируемые модули

from flask import Blueprint, jsonify, request, url_for
from flask_login import login_user, logout_user
from flask_mail import Message

from extensions import db, mail
from models import User

• Blueprint: Используется для создания модуля аутентификации.

• jsonify: Функция для преобразования данных в JSON-формат.

• request: Объект, представляющий текущий HTTP-запрос.

• url_for: Функция для генерации URL-адресов по имени маршрута.

• login_user, logout_user: Функции для управления сессиями пользователей.

• Message: Класс для создания и отправки электронных писем.


Инициализация Blueprint

auth_bp = Blueprint("main", __name__)

• Создается объект Blueprint с именем "auth", который будет группировать маршруты, относящиеся к аутентификации.


Маршрут /login

@auth_bp.route("/login", methods=["POST"])
def login():
    data = request.json
    user = User.query.filter_by(email=data["email"]).first()
    if user and user.check_password(data["password"]):
        login_user(user)
        return jsonify({"message": "Login successful"}), 200
    return jsonify({"error": "Invalid credentials"}), 401

• Этот маршрут обрабатывает POST-запросы для входа пользователя в систему. Сначала происходит проверка существования пользователя по электронной почте, а затем проверяется правильность введенного пароля. В случае успешной аутентификации, пользователь будет залогинен и получит сообщение об успешном входе. Если данные некорректны, возвращается ошибка 401.


Маршрут /logout

@auth_bp.route("/logout", methods=["POST"])
def logout():
    logout_user()
    return jsonify({"message": "Logout successful"}), 200

• Этот маршрут обрабатывает POST-запросы для выхода пользователя из системы. Пользователь выйдет из аккаунта, и на ответ будет отправлено сообщение об успешном выходе.


Маршрут /reset_password

@auth_bp.route("/reset_password", methods=["POST"])
def reset_password():
    data = request.json
    user = User.query.filter_by(email=data["email"]).first()
    if not user:
        return jsonify({"error": "User not found"}), 404

    token = user.id  # Using ID as an example token
    reset_url = url_for("auth.complete_reset", token=token, _external=True)

    msg = Message("Password Reset Request", recipients=[user.email])
    msg.body = f"To reset your password, visit the following link: {reset_url}"
    mail.send(msg)

    return jsonify({"message": "Reset link sent"}), 200

• Этот маршрут обрабатывает запросы на сброс пароля. На основе переданного адреса электронной почты ищется пользователь. Если пользователь не найден, возвращается ошибка 404. Если пользователь существует, создается временный токен (в данном случае используется идентификатор пользователя) и формируется URL для сброса пароля. На указанный адрес электронной почты отправляется сообщение с инструкцией по сбросу пароля.


Маршрут /reset_password/

@auth_bp.route("/reset_password/<token>", methods=["POST"])
def complete_reset(token):
    user = User.query.get(int(token))
    if not user:
        return jsonify({"error": "Invalid token"}), 400

    data = request.json
    user.set_password(data["password"])
    db.session.commit()
    return jsonify({"message": "Password reset successful"}), 200

• Этот маршрут обрабатывает запросы для завершения процесса сброса пароля. Проверяется корректность токена (идентификатора пользователя). Если токен некорректный, возвращается ошибка 400. В случае корректного токена, новый пароль устанавливается для пользователя, и изменения сохраняются в базе данных.

⚠️ **GitHub.com Fallback** ⚠️