Build a Flask‑Elasticsearch Search Engine: From Config to Deployment

This article demonstrates how to build a search engine using Flask and Elasticsearch, providing step‑by‑step code and configuration.

Configuration File

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

Logging Configuration

# coding=utf-8
import os, logging, logging.config as log_conf, datetime, 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)

Blueprints and Routes

# 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)

# 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
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

# -*- coding:utf8 -*-
from flask import Blueprint
@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', 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(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

# 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()

Run the project with python manage.py runserver; the default address is http://127.0.0.1:5000.

Gunicorn Deployment

# 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'

Start Gunicorn with the configuration file: gunicorn -c gconfig.py manage:app.

Project source code is available at https://github.com/GZKY-PY/Flask-ES.