This repository has been archived on 2020-09-13. You can view files and clone it, but cannot push or open issues or pull requests.
magic-app/magic_app/app.py
2020-07-05 02:41:06 +02:00

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