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

Configuration File

The Config.py module defines database connection parameters (optional for this project) and a Config class that sets Flask secret key, SQLAlchemy options, and mail server settings.

#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 = "random_string"
    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 = 'mail_auth_code'
    FLASK_MAIL_SUBJECT_PREFIX = 'M_KEPLER'
    FLASK_MAIL_SENDER = MAIL_USERNAME
    MAIL_USE_TLS = False
    MAIL_DEBUG = False
    ENABLE_THREADS = True

Logger

A robust logging configuration is provided in Logger.py. It creates a log directory, defines console and rotating file handlers, and uses coloredlogs for colored console output.

# 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 application uses Flask blueprints to separate the Math and Baike sections. The main app initialization registers these blueprints and sets up SQLAlchemy, CSRF protection, and mail support.

# -*- coding:utf8 -*-
from flask import Flask
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)

# Import blueprints after db is created
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")

Blueprint definitions:

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

# -*- coding:utf-8 -*-
from flask import Blueprint
math = Blueprint("math", __name__)
from app.home.math import views

Example routes for the Baike section demonstrate search handling, pagination with flask_paginate, logging of search keywords, and dynamic rendering of detail pages based on document IDs.

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

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(get_page_parameter(), type=int, default=1)
        start = (page - 1) * PER_PAGE
        end = start + PER_PAGE
        total = 30
        pagination = Pagination(page=page, start=start, end=end, total=total)
        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('app/templates/s_d/{}.html'.format(uid))
    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('s_d/{}.html'.format(uid), match_data=match_data)

Project Startup

The project uses flask_script to provide a convenient runserver command. 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

For production, Gunicorn is recommended. Install it with pip install gunicorn, configure workers, threads, and logging in gconfig.py, then launch 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'

Finally, the source code is hosted on GitHub at https://github.com/GZKY-PY/Flask-ES .