주기적으로 요청해야하는 프로그램이 있어 스케쥴러 프로그램을 작성했더니 (flask)
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask
from .batch import check_not_answer_question
from .database import db
from . import config
from sqlalchemy import text
from .routes import routes_bp
global_api_key = config.API_KEY
def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = config.DATABASE_URI
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)
# 데이터베이스 연결 테스트
with app.app_context():
try:
# 데이터베이스 엔진으로부터 커넥션을 얻고 SELECT 1을 실행
with db.engine.connect() as connection:
result = connection.execute(text("SELECT 1"))
for row in result:
print("Connection test returned:", row[0])
print("Database connected successfully.")
except Exception as e:
print("Failed to connect to database.")
print(e)
# 라우트 등록
from . import routes
# 스케쥴러 등록
scheduler = BackgroundScheduler()
scheduler.add_job(func=check_not_answer_question, trigger='interval', seconds=10)
scheduler.start()
app.register_blueprint(routes_bp)
return app스케쥴러에 등록하려는check_not_answer_question 에서 db의 models를 참조하고있기에 아직 app 객체가 생성되기전에 참조하려는 문제가 일어나 아래와같은 오류가 발생했다.
Error: While importing 'app.run', an ImportError was raised:
Traceback (most recent call last):
File "D:\py\tmt-ai-server\venv\lib\site-packages\flask\cli.py", line 247, in locate_app
__import__(module_name)
File "D:\py\tmt-ai-server\app\__init__.py", line 4, in <module>
from .batch import check_not_answer_question
File "D:\py\tmt-ai-server\app\batch.py", line 3, in <module>
from app.models import JobApplicationQuestions, UserJobApplicationInfo
File "D:\py\tmt-ai-server\app\models.py", line 3, in <module>
from . import db
ImportError: cannot import name 'db' from partially initialized module 'app' (most likely due to a circular import) (D:\py\tmt-ai-server\app\__init__.py)
Process finished with exit code 2
python도 자주쓰는 언어가 아니고 flask도 어떻게 코드를 설계하는지 몰라 설계를 대충했더니 생긴 문제다..
import문의 위치도 바꿔주고 지연참조로 수정해주니 배치가 실행된다.
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask
from .database import db
from . import config
from sqlalchemy import text
from .routes import routes_bp
from .batch import check_not_answer_question
global_api_key = config.API_KEY
def create_app():
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = config.DATABASE_URI
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.init_app(app)
# 데이터베이스 연결 테스트
with app.app_context():
try:
# 데이터베이스 엔진으로부터 커넥션을 얻고 SELECT 1을 실행
with db.engine.connect() as connection:
result = connection.execute(text("SELECT 1"))
for row in result:
print("Connection test returned:", row[0])
print("Database connected successfully.")
except Exception as e:
print("Failed to connect to database.")
print(e)
# 라우트 등록
from . import routes
app.register_blueprint(routes_bp)
# 스케쥴러 설정 및 작업 등록을 위한 함수 정의
def setup_scheduler():
scheduler = BackgroundScheduler()
# app_context를 사용하여 함수 내에서 애플리케이션 컨텍스트를 사용
def job_function():
with app.app_context():
from .batch import check_not_answer_question
check_not_answer_question()
scheduler.add_job(func=check_not_answer_question, trigger='interval', seconds=10)
scheduler.start()
# 애플리케이션 컨텍스트가 설정된 후에 스케쥴러 설정
setup_scheduler()
return app
이제 물론 배치에서도 또다른 에러가 나긴하지만.. 하나를 해결한데에 의의를 두자
-- 답변되지 않은 값들을 확인 후 답변 요청하는 배치 프로그램 실행 --
-- 유저확인 중 --
배치 작업 중 오류 발생: Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.
스케쥴링 코드를 짜는데 너무 알아야할 지식도 많고 그냥 60초마다 실행되는 코드를 짜면안되나했더니 Flask의 개발서버는 보통 싱글스레드로 동작하기에.. app.run()하고 while문으로 실행할시 메인스레드를 계속 차지하여 다른요청을 처리하지 못한다고한다.
-> 하루지난 오늘 해결했다! 결국 파이썬은 flask서버를 켜지않아도 실행가능하기에.. 스케쥴링 코드를 flask 컨텍스트에서 실행되게 해야하는데 이전 코드들은 컨텍스트에서 실행한다는 코드를 넣지않았기에 application context밖에서 실행되고있다는 오류메세지가 뜬거였다.
run.py에 이런식으로 내가 실행하고자하는 (DB관련된 스케쥴링 코드) 함수를 app.app_context()에서 실행할거라고 작성하면 app의 context에서 실행된다.
# run.py
from flask_apscheduler import APScheduler
from app import create_app
from app.batch import check_not_answer_question
class Config:
SCHEDULER_API_ENABLED = True
app = create_app()
app.config.from_object(Config())
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()
# 스케쥴러 작업 정의
@scheduler.task('interval', id='schedule_1', seconds=5, misfire_grace_time=900)
def schedule_1():
# 5초 단위로 실행하니까 이후에 단위 수정필요
with app.app_context(): # 애플리케이션 컨텍스트 설정
print("Schedule Function is executed.")
check_not_answer_question()
if __name__ == '__main__':
app.run()
잘 실행된다!
댓글