Flask SQLAlchemy 세션이 동기화되지 않았습니다.

Question

gunicorn/nginx 스택과 함께 실행되는 Flask REST API가 있습니다. API가 실행되는 각 스레드마다 한 번 설정된 전역 SQLAlchemy 세션이 있습니다. 필자는 API의 단위 테스트를 실행하기 위해 endpoint/test /를 설정했습니다. 정리 절 : 하나 개의 테스트 데이터베이스에 무언가를 추가하는 POST 요청을 만들고, 다음 마지막을 가지고

def test_something(): 
    try: 
     url = "http://myposturl" 
     data = {"content" : "test post"} 
     headers = {'content-type': 'application/json'} 
     result = requests.post(url, json=data, headers=headers).json() 
     validate(result, myschema) 
    finally: 
     db.sqlsession.query(MyTable).filter(MyTable.content == "test post").delete() 
     db.sqlsession.commit() 

문제는 그 POST 요청이 이제 만든 "테스트 포스트"가되는 스레드 개체가 세션에 있지만 데이터베이스에 이러한 개체가 없습니다. 테스트가 실행 된 스레드가 데이터베이스에서 해당 개체를 삭제했기 때문입니다. 그래서 서버에 GET 요청을하면 약 4 번 (4 명의 gunicorn 작업자가 있음), "테스트 포스트"개체가 생기고 4 번은 3 번이 나옵니다. 이것은 스레드가 각각 자신의 세션 개체를 가지고 있기 때문에 동기를 잃고 있지만 실제로 그것에 대해 어떻게해야할지 모르겠다. ...

다음은 SQLAlchemy 세션에 대한 나의 설정이다 :

def connectSQLAlchemy(): 
    import sqlalchemy 
    import sqlalchemy.orm 
    engine = sqlalchemy.create_engine(connection_string(DBConfig.USER, DBConfig.PASSWORD, DBConfig.HOST, DBConfig.DB)) 
    session_factory = sqlalchemy.orm.sessionmaker(bind=engine) 
    Session = sqlalchemy.orm.scoped_session(session_factory) 
    return Session() 

# Create a global session for everyone 
sqlsession = connectSQLAlchemy() 

Answer 1

플라스크를 사용하는 경우 flask-sqlalchemy를 사용하십시오. 플라스크를 사용하는 경우 세션의 수명주기를 처리합니다.

직접 해보려는 경우 올바른 패턴은 글로벌 세션이 아니라 각 요청에 대한 세션을 만드는 것입니다. 대신

Session = scoped_session(session_factory) 
return Session() 

의

Session = scoped_session(session_factory, scopefunc=flask._app_ctx_stack.__ident_func__) 
return Session 

을하고

session = Session() 

세션을 필요로 할 때마다 수행해야합니다. scoped_session과 scopefunc 덕분에 각 요청에서 다른 세션이 반환되지만 동일한 요청에서 동일한 세션이 반환됩니다.

Answer 2

알아 냈어. 여전히 내 DB 모듈의 글로벌 세션 객체를 사용

이

@app.before_request 
def startup_session(): 
    db.session = db.connectSQLAlchemy() 

@app.teardown_request 
def shutdown_session(exception=None): 
    db.session.close() 

:

db.py: 

.... 
session = None 
.... 

scoped_session는 다른 처리 내가 한 일은 내 응용 프로그램의 __init__.py의 요청에 대한 설치 및 해체를 추가했다 스레드, 내 생각 엔 ...

어떤 이유로이 일을하는 것이 끔찍한 방법인지 알려주십시오. = c)