75 lines
1.6 KiB
Python
75 lines
1.6 KiB
Python
"""The main application factory."""
|
|
from logging import DEBUG, ERROR, getLogger
|
|
|
|
from celery import Celery
|
|
from flask import Flask
|
|
from redis import Redis
|
|
|
|
from magic_app.config import Base, Production
|
|
|
|
celery = Celery(
|
|
__name__, backend=Base.CELERY_RESULT_BACKEND, broker=Base.CELERY_BROKER_URL
|
|
)
|
|
|
|
|
|
def create_app(config=Production):
|
|
"""Main application factory."""
|
|
app = Flask(__name__.split(".")[0])
|
|
app.config.from_object(config)
|
|
|
|
configure_celery(app)
|
|
configure_redis(app)
|
|
configure_views(app)
|
|
configure_logging(app)
|
|
|
|
return app
|
|
|
|
|
|
def configure_redis(app):
|
|
"""Configure Redis connection."""
|
|
from magic_app.session import RedisSessionDB
|
|
|
|
host = Base.REDIS_HOST
|
|
port = Base.REDIS_PORT
|
|
db_num = Base.REDIS_SESSION_DB
|
|
|
|
if app.testing:
|
|
from fakeredis import FakeRedis
|
|
|
|
connection = FakeRedis()
|
|
else:
|
|
connection = Redis(host=host, port=port, db=db_num)
|
|
|
|
app.config["SESSION"] = RedisSessionDB(connection)
|
|
|
|
|
|
def configure_celery(app):
|
|
"""Configure celery."""
|
|
celery.conf.update(app.config)
|
|
|
|
class ContextTask(celery.Task):
|
|
def __call__(self, *args, **kwargs):
|
|
with app.app_context():
|
|
return self.run(*args, **kwargs)
|
|
|
|
celery.Task = ContextTask
|
|
|
|
|
|
def configure_views(app):
|
|
"""Configure API resource views."""
|
|
from magic_app.views import apps
|
|
|
|
app.register_blueprint(apps)
|
|
|
|
|
|
def configure_logging(app):
|
|
"""Configure application logging."""
|
|
app_logger = getLogger(__name__)
|
|
|
|
if not app.debug:
|
|
app_logger.setLevel(ERROR)
|
|
else:
|
|
app_logger.setLevel(DEBUG)
|
|
|
|
app.logging = app_logger
|