I have taken over a flask app, but it does not use the flask-sqlalchemy plugin. I am having a hard time wrapping my head around how it's set up.
It has a database.py
file.
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session, Session
_session_factory = None
_scoped_session_cls = None
_db_session: Session = None
def _get_session_factory():
global _session_factory
if _session_factory is None:
_session_factory = sessionmaker(
bind=create_engine(CONNECTION_URL)
)
return _session_factory
def new_session():
session_factory = _get_session_factory()
return session_factory()
def new_scoped_session():
global _scoped_session_cls
if _scoped_session_cls is None:
session_factory = _get_session_factory()
if not session_factory:
return
_scoped_session_cls = scoped_session(session_factory)
return _scoped_session_cls()
def init_session():
global _db_session
if _db_session is not None:
log.warning("already init")
else:
_db_session = new_scoped_session()
return _db_session
def get_session():
return _db_session
We we start up the flask app, it calls database.init_session()
and then anytime we want to use the database it calls database.get_session()
.
Is this a correct/safe way to interact with the database? What happens if there are two requests being processed at the same time by different threads? Will this result in cross-contamination with both using the same session
from SQLAlchemy, Flask and cross-contamination?
No comments:
Post a Comment