Build a Flask‑Elasticsearch Search Engine: Config, Logging, Routing & Deployment Guide

Configuration File

The Config.py file defines database connection parameters, secret key, SQLAlchemy settings, mail server configuration, and other Flask extensions. Although the project uses MySQL for auxiliary purposes, Elasticsearch is sufficient for data storage, and email notifications are optional.

#coding:utf-8
import os
DB_USERNAME = 'root'
DB_PASSWORD = None  # If no password
DB_HOST = '127.0.0.1'
DB_PORT = '3306'
DB_NAME = 'flask_es'

class Config:
    SECRET_KEY = "RandomString"
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://%s:%s@%s:%s/%s' % (DB_USERNAME, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME)
    MAIL_SERVER = 'smtp.qq.com'
    MAIL_POST = 465
    MAIL_USERNAME = '[email protected]'
    MAIL_PASSWORD = 'EmailAuthCode'
    FLASK_MAIL_SUBJECT_PREFIX = 'M_KEPLER'
    FLASK_MAIL_SENDER = MAIL_USERNAME
    MAIL_USE_TLS = False
    MAIL_DEBUG = False
    ENABLE_THREADS = True

Logger

A reusable logging configuration is provided in Logger.py. It sets up colored console output, rotating file handlers, and defines loggers for different levels. The configuration writes logs to a logs directory, creating it if necessary.

# coding=utf-8
import os
import logging
import logging.config as log_conf
import datetime
import coloredlogs

coloredlogs.DEFAULT_FIELD_STYLES = {
    'asctime': {'color': 'green'},
    'hostname': {'color': 'magenta'},
    'levelname': {'color': 'magenta', 'bold': False},
    'name': {'color': 'green'}
}

log_dir = os.path.dirname(os.path.dirname(__file__)) + '/logs'
if not os.path.exists(log_dir):
    os.mkdir(log_dir)

today = datetime.datetime.now().strftime("%Y-%m-%d")
log_path = os.path.join(log_dir, today + ".log")

log_config = {
    'version': 1.0,
    'formatters': {
        'colored_console': {
            'format': "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            'datefmt': '%H:%M:%S'
        },
        'detail': {
            'format': "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            'datefmt': "%Y-%m-%d %H:%M:%S"
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'level': 'DEBUG',
            'formatter': 'colored_console'
        },
        'file': {
            'class': 'logging.handlers.RotatingFileHandler',
            'maxBytes': 1024*1024*1024,
            'backupCount': 1,
            'filename': log_path,
            'level': 'INFO',
            'formatter': 'detail',
            'encoding': 'utf-8'
        }
    },
    'loggers': {
        'logger': {
            'handlers': ['console'],
            'level': 'DEBUG'
        }
    }
}

log_conf.dictConfig(log_config)
log_v = logging.getLogger('log')
coloredlogs.install(level='DEBUG', logger=log_v)

Routes and Blueprints

The project uses Flask blueprints to separate the Math and Baike sections. Pagination is handled by flask_paginate, and search queries are sent to Elasticsearch. Each detail page is generated dynamically, ensuring only one HTML template exists per UID.

# -*- coding:utf-8 -*-
from flask import Blueprint
baike = Blueprint("baike", __name__)

# baike/views.py would contain route implementations

# -*- coding:utf8 -*-
from flask import Blueprint
math = Blueprint("math", __name__)

# -*- coding:utf8 -*-
from flask import Flask, request, render_template, redirect
from flask_sqlalchemy import SQLAlchemy
from app.config.config import Config
from flask_mail import Mail
from flask_wtf.csrf import CSRFProtect

app = Flask(__name__, template_folder='templates', static_folder='static')
app.config.from_object(Config)

db = SQLAlchemy(app)
csrf = CSRFProtect(app)
mail = Mail(app)

# Register blueprints
from app.home.baike import baike as baike_blueprint
from app.home.math import math as math_blueprint
from app.home.home import home as home_blueprint

app.register_blueprint(home_blueprint)
app.register_blueprint(math_blueprint, url_prefix="/math")
app.register_blueprint(baike_blueprint, url_prefix="/baike")

# -*- coding:utf-8 -*-
from flask import Blueprint, request, render_template, redirect
from app.Logger.logger import log_v
from app.elasticsearchClass import elasticSearch
from app.home.forms import SearchForm

baike_es = elasticSearch(index_type="baike_data", index_name="baike")

@baike.route('/')
def index():
    searchForm = SearchForm()
    return render_template('baike/index.html', searchForm=searchForm)

@baike.route('/search', methods=['GET', 'POST'])
def baikeSearch():
    search_key = request.args.get('b', default=None)
    if search_key:
        searchForm = SearchForm()
        log_v.error("[+] Search Keyword: " + search_key)
        match_data = baike_es.search(search_key, count=30)
        PER_PAGE = 10
        page = request.args.get('page', type=int, default=1)
        start = (page - 1) * PER_PAGE
        end = start + PER_PAGE
        pagination = Pagination(page=page, start=start, end=end, total=30)
        context = {
            'match_data': match_data["hits"]["hits"][start:end],
            'pagination': pagination,
            'uid_link': '/baike/'
        }
        return render_template('data.html', q=search_key, searchForm=searchForm, **context)
    return redirect('home.index')

@baike.route('/<uid>')
def baikeSd(uid):
    base_path = os.path.abspath('app/templates/s_d/')
    old_file = os.listdir(base_path)[0]
    old_path = os.path.join(base_path, old_file)
    file_path = os.path.abspath(f'app/templates/s_d/{uid}.html')
    if not os.path.exists(file_path):
        log_v.debug("[-] File does not exist, renaming !!!")
        os.rename(old_path, file_path)
    match_data = baike_es.id_get_doc(uid=uid)
    return render_template(f's_d/{uid}.html', match_data=match_data)

Project Startup

The application is launched with flask_script. Running python manage.py runserver starts the development server on port 5000.

# coding:utf8
from app import app
from flask_script import Manager, Server

manage = Manager(app)
manage.add_command("runserver", Server(use_debugger=True))

if __name__ == "__main__":
    manage.run()

Gunicorn Deployment

Gunicorn can be installed via pip install gunicorn. A separate configuration file (e.g., gconfig.py) defines workers, threads, binding address, log files, and other options. The server is started with gunicorn -c gconfig.py manage:app.

# encoding:utf-8
import multiprocessing
from gevent import monkey
monkey.patch_all()

workers = multiprocessing.cpu_count() * 2 + 1
debug = True
reload = True
loglevel = 'debug'
threads = 2
bind = '0.0.0.0:5001'
daemon = 'false'
worker_class = 'gevent'
worker_connections = 2000
pidfile = 'log/gunicorn.pid'
logfile = 'log/debug.log'
accesslog = 'log/gunicorn_acess.log'
errorlog = 'log/gunicorn_error.log'

Running the Application

After configuring the environment, start the Flask development server or launch Gunicorn for production. The application will be accessible at http://127.0.0.1:5000 (or the port specified in the Gunicorn config).