Как гарантировать, что работники сельдерея имеют доступ к тому же файлу .env, что и flask? ⇐ Python

[Anonymous]

Я работаю над сервером Flask и пытаюсь интегрировать celery для обработки преобразования данных, поступающих через веб-перехватчик. Я широко использую переменные среды в настройке моей колбы для таких вещей, как ключи API и доступ к базе данных, и все это хранится в файле .env. Я хотел бы иметь экземпляр flask в работнике сельдерея, чтобы обеспечить доступ к моделям sqlalchemy, объекту db и т. д., но когда я пытаюсь создать работника, фабрика приложений для Flask завершается с ошибкой, потому что переменные .env нигде не встречаются. Есть ли способ предварительно загрузить их в Flask перед попыткой инициализации базы данных?
Выполнение команды

Код: Выделить всё

celery -A make_celery worker --loglevel INFO

результат

Код: Выделить всё

RuntimeError: Either 'SQLALCHEMY_DATABASE_URI' or 'SQLALCHEMY_BINDS' must be set.

Flask может нормально работать сам по себе, и мне сложно отлаживать рабочий процесс, потому что он не выводит на консоль.
Мой файл make_celery выглядит так:

Код: Выделить всё

import os

from app import create_app

flask_app = create_app(os.getenv('FLASK_CONFIG') or 'default')
celery_app = flask_app.extensions["celery"]

и метод create_app (вроде анонимизированный) такой:

Код: Выделить всё

from celery_app import celery_init_app
from config import config

db = SQLAlchemy()

def create_app(config_name):
app = Flask(__name__)
app.config.from_object(config[config_name])
config[config_name].init_app(app)

# worker fails here every time as SQLAlCHEMY_DATABASE_URI is not set
db.init_app(app)

with app.app_context():
#not included is importing of models and migrate init

app.config.from_mapping(
CELERY=dict(
broker_url="amqp://guest:guest@127.0.0.1:5672//",
result_backend=app.config['CELERY_BACKEND_URL'],
task_ignore_result=True,
),
)
app.config.from_prefixed_env()
celery_init_app(app)

return app

И метод celery_init_app(), включенный в метод create_app():

Код: Выделить всё

def celery_init_app(app: Flask) -> Celery:
class FlaskTask(Task):
def __call__(self, *args: object, **kwargs: object) -> object:
with app.app_context():
return self.run(*args, **kwargs)

celery_app = Celery(app.name, task_cls=FlaskTask)
celery_app.config_from_object(app.config["CELERY"])
celery_app.set_default()
app.extensions["celery"] = celery_app
return celery_app

Я новичок в Flask и Celery и не могу взломать этот.
Я запустил приведенную выше рабочую команду celery, ожидая, что она создайте работника, который будет иметь доступ к экземпляру приложения flask и всем его компонентам, но вместо этого ему не удалось инициировать объект базы данных sqlalchemy из-за отсутствия переменных среды.

Подробнее здесь: https://stackoverflow.com/questions/790 ... e-as-flask