Skip to content
Snippets Groups Projects

use flask-sqlalchemy for the main db

Merged Remco Tukker requested to merge feature/COMP-170_part_2_introduce_flask_sqlalchemy into develop
1 unresolved thread
Files
26
+ 11
39
import contextlib
import logging
from typing import Optional, Union, Callable, Iterator
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import sessionmaker, Session
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import MetaData
logger = logging.getLogger(__name__)
_SESSION_MAKER: Union[None, sessionmaker] = None
@contextlib.contextmanager
def session_scope(
callback_before_close: Optional[Callable] = None) -> Iterator[Session]:
# best practice is to keep session scope separate from data processing
# cf. https://docs.sqlalchemy.org/en/13/orm/session_basics.html
assert _SESSION_MAKER
session = _SESSION_MAKER()
try:
yield session
session.commit()
if callback_before_close:
callback_before_close()
except SQLAlchemyError:
logger.error('caught sql layer exception, rolling back')
session.rollback()
raise # re-raise, will be handled by main consumer
finally:
session.close()
def postgresql_dsn(db_username, db_password, db_hostname, db_name, port=5432):
return (f'postgresql://{db_username}:{db_password}'
f'@{db_hostname}:{port}/{db_name}')
logger = logging.getLogger(__name__)
def init_db_model(dsn):
global _SESSION_MAKER
metadata_obj = MetaData(naming_convention={
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
})
# cf. https://docs.sqlalchemy.org/en
# /latest/orm/extensions/automap.html
engine = create_engine(dsn, pool_size=10)
_SESSION_MAKER = sessionmaker(bind=engine)
db = SQLAlchemy(metadata=metadata_obj)
Loading