| |
| from __future__ import annotations |
| from datetime import datetime, timedelta, timezone |
| from typing import TYPE_CHECKING, Any |
|
|
| from flask import Flask, abort, redirect, request, url_for |
| from flask_admin import Admin, AdminIndexView, expose, helpers |
| from flask_admin.consts import ICON_TYPE_FONT_AWESOME |
| from flask_admin.contrib.sqla import ModelView |
| from flask_babel import Babel |
| from flask_caching import Cache |
| from flask_login import current_user |
| from flask_security.core import RoleMixin, Security, UserMixin |
| from flask_security.datastore import SQLAlchemyUserDatastore |
| from flask_security.utils import hash_password |
| from flask_sqlalchemy import SQLAlchemy |
| from sqlalchemy import inspect |
| from wtforms import PasswordField |
|
|
| from admin.views.users import UserView as AppUserView |
| from bot.database.models import UserModel as AppUserModel |
|
|
| if TYPE_CHECKING: |
| from werkzeug.wrappers.response import Response |
|
|
| |
| app = Flask(__name__) |
| app.config.from_pyfile("config.py") |
| db = SQLAlchemy(app) |
| cache = Cache(app) |
| babel = Babel(app) |
|
|
| |
| roles_admins = db.Table( |
| "roles_admins", |
| db.Column("admin_id", db.Integer(), db.ForeignKey("admin.id")), |
| db.Column("role_id", db.Integer(), db.ForeignKey("role.id")), |
| ) |
|
|
|
|
| class RoleModel(db.Model, RoleMixin): |
| __tablename__ = "role" |
|
|
| id = db.Column(db.Integer(), primary_key=True) |
| name = db.Column(db.String(80), unique=True) |
| description = db.Column(db.String(255)) |
|
|
| def __str__(self) -> str: |
| return self.name |
|
|
|
|
| class AdminModel(db.Model, UserMixin): |
| __tablename__ = "admin" |
|
|
| id = db.Column(db.Integer, primary_key=True) |
| first_name = db.Column(db.String(255)) |
| last_name = db.Column(db.String(255)) |
| email = db.Column(db.String(255), unique=True, nullable=False) |
| password = db.Column(db.String(255), nullable=False) |
| active = db.Column(db.Boolean()) |
| confirmed_at = db.Column(db.DateTime(), default=datetime.utcnow) |
| fs_uniquifier = db.Column(db.String(255), unique=True) |
| roles = db.relationship("RoleModel", secondary=roles_admins, backref=db.backref("admins", lazy="dynamic")) |
|
|
| def __str__(self) -> str: |
| return self.email |
|
|
|
|
| |
| admin_datastore = SQLAlchemyUserDatastore(db, AdminModel, RoleModel) |
| security = Security(app, admin_datastore) |
|
|
|
|
| |
| class RoleView(ModelView): |
| can_delete = False |
| can_edit = False |
| can_create = False |
| can_view_details = False |
| edit_modal = True |
| create_modal = True |
| can_export = False |
| details_modal = True |
|
|
| def is_accessible(self) -> bool: |
| if not current_user.is_active or not current_user.is_authenticated: |
| return False |
|
|
| return bool(current_user.has_role("superuser")) |
|
|
| def _handle_view(self, _name: str, **_kwargs: dict) -> Response | None: |
| """Override builtin _handle_view in order to redirect users when a view is not accessible.""" |
| if not self.is_accessible(): |
| if current_user.is_authenticated: |
| |
| abort(403) |
| else: |
| |
| return redirect(url_for("security.login", next=request.url)) |
| return None |
|
|
|
|
| class AdminView(RoleView): |
| can_view_details = True |
| can_delete = True |
| can_edit = True |
| can_export = True |
| can_create = True |
| export_types = ["csv", "xlsx", "json", "yaml"] |
|
|
| column_editable_list = ["email", "first_name", "last_name"] |
| column_searchable_list = column_editable_list |
| column_exclude_list = ["password"] |
| form_excluded_columns = ["confirmed_at"] |
| column_details_exclude_list = column_exclude_list |
| column_filters = column_editable_list |
| form_overrides = {"password": PasswordField} |
|
|
|
|
| |
| def get_orders_count() -> int: |
| return 0 |
|
|
|
|
| def get_user_count() -> int: |
| return db.session.query(AppUserModel).count() |
|
|
|
|
| def get_new_user_count(days_before: int = 1) -> int: |
| period_start = datetime.now(timezone.utc) - timedelta(days=days_before) |
| return db.session.query(AppUserModel).filter(AppUserModel.created_at >= period_start).count() |
|
|
|
|
| class CustomAdminIndexView(AdminIndexView): |
| @expose("/") |
| def index(self) -> str: |
| days_before: int = 1 |
| period_start = datetime.now(timezone.utc) - timedelta(days=days_before) |
| order_count = get_orders_count() |
| user_count = get_user_count() |
| new_user_count = get_new_user_count(days_before) |
| new_user_count = get_new_user_count(days_before) |
|
|
| return self.render( |
| "admin/index.html", |
| order_count=order_count, |
| user_count=user_count, |
| new_user_count=new_user_count, |
| period_start=period_start, |
| default_email=app.config.get("DEFAULT_ADMIN_EMAIL"), |
| default_password=app.config.get("DEFAULT_ADMIN_PASSWORD"), |
| ) |
|
|
|
|
| @app.route("/") |
| def index() -> Response: |
| return redirect(url_for("admin.index")) |
|
|
|
|
| |
| admin = Admin( |
| app, |
| name="Telegram Bot", |
| base_template="my_master.html", |
| index_view=CustomAdminIndexView( |
| name="Home", |
| url="/admin", |
| menu_icon_type=ICON_TYPE_FONT_AWESOME, |
| menu_icon_value="fa-home", |
| ), |
| template_mode="bootstrap4", |
| ) |
|
|
| admin.add_view( |
| AppUserView( |
| AppUserModel, |
| db.session, |
| menu_icon_type=ICON_TYPE_FONT_AWESOME, |
| menu_icon_value="fa-users", |
| name="Users", |
| endpoint="users", |
| ), |
| ) |
|
|
| admin.add_view( |
| AdminView( |
| AdminModel, |
| db.session, |
| menu_icon_type=ICON_TYPE_FONT_AWESOME, |
| menu_icon_value="fa-black-tie", |
| name="Admins", |
| endpoint="admins", |
| ), |
| ) |
| admin.add_view( |
| RoleView( |
| RoleModel, |
| db.session, |
| menu_icon_type=ICON_TYPE_FONT_AWESOME, |
| menu_icon_value="fa-tags", |
| name="Roles", |
| endpoint="roles", |
| ), |
| ) |
|
|
|
|
| |
| @security.context_processor |
| def security_context_processor() -> dict[str, Any]: |
| return { |
| "admin_base_template": admin.base_template, |
| "admin_view": admin.index_view, |
| "h": helpers, |
| "get_url": url_for, |
| } |
|
|
|
|
| def init_db() -> None: |
| inspector = inspect(db.engine) |
| if inspector.has_table("admin") and inspector.has_table("role"): |
| return |
|
|
| db.create_all() |
|
|
| admin_role = RoleModel(name="user", description="does not have access to other administrators") |
| super_admin_role = RoleModel(name="superuser", description="has access to manage all administrators") |
| db.session.add(admin_role) |
| db.session.add(super_admin_role) |
| db.session.commit() |
|
|
| admin_datastore.create_user( |
| first_name="Admin", |
| email=app.config.get("DEFAULT_ADMIN_EMAIL"), |
| password=hash_password(str(app.config.get("DEFAULT_ADMIN_PASSWORD"))), |
| roles=[admin_role, super_admin_role], |
| ) |
|
|
| db.session.commit() |
|
|
| return |
|
|
|
|
| with app.app_context(): |
| init_db() |
|
|
| if __name__ == "__main__": |
| app.run(host=app.config.get("ADMIN_HOST"), port=app.config.get("ADMIN_PORT"), debug=app.config.get("DEBUG")) |
|
|