mew/core/db.py

from contextlib import contextmanager

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.scoping import scoped_session

from core import constants

engine = create_engine(
    f'sqlite:///{constants.DB_PATH}',
    connect_args={'check_same_thread': False},
    echo=constants.SQL_ALCHEMY_VERBOSE)
Base = declarative_base()

# session_factory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# session = scoped_session(session_factory)
# # session_ = Session()
# #
# #


def session():
    return Session(engine)


@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()


def create():
    if constants.DB_PATH.exists():
        raise FileExistsError('A db file already exists')
    Base.metadata.create_all(engine)


def close():
    # session.remove()
    engine.dispose()