sess

``rattail.db.sess``

===================

.. automodule:: rattail.db.sess

   :members:

   .. class:: Session

      SQLAlchemy session class used for all (normal) :term:`app database`

      connections.

      This is a subclass of :class:`SessionBase`.

      See the :class:`sqlalchemy:sqlalchemy.orm.Session` docs for more

      info.

      .. note::

         WuttJamaican also provides a

         :class:`~wuttjamaican:wuttjamaican.db.sess.Session` class,

         however for now Rattail still must define/use its own.

"rattail" = "rattail.config:RattailConfigExtension"

        # TODO: deprecate / remove these

        if hasattr(self, 'appdb_engines'):

            self.rattail_engines = self.appdb_engines

            self.rattail_engine = self.appdb_engine

        # configure session if applicable

        from rattail.db import Session

        if Session:

            # nb. wutta configures its own Session but we must

            # configure ours separately (albeit with same engine)

            if hasattr(self, 'appdb_engine'):

                Session.configure(bind=self.appdb_engine)

            # TODO: eventually this should not be needed (?)

            Session.configure(rattail_config=self)

            # TODO: this should be removed, it sets 'record changes' globally

            from rattail.db.config import configure_session

            configure_session(self, Session)

    DEPRECATED - use

    :class:`wuttjamaican:wuttjamaican.conf.WuttaConfigExtension`

    instead.

    See also :class:`RattailConfigExtension`.

    def __init__(self, *args, **kwargs):

        super().__init__(*args, **kwargs)

        warnings.warn(f"rattail.config.ConfigExtension (for '{self.key}') is deprecated; "

                      "please use wuttjamaican.conf.ConfigExtension instead",

                      DeprecationWarning, stacklevel=2)

class RattailConfigExtension(WuttaConfigExtension):

    """

    The :term:`config extension` for Rattail.

    This is a subclass of

    :class:`wuttjamaican:wuttjamaican.conf.WuttaConfigExtension`.

    This primarily exists to add default config settings for sake of

    the various importers etc. contained in the ``rattail`` package.

    .. note::

       If you are adding a new config extension, **do not subclass

       this**, but rather subclass

       :class:`wuttjamaican:wuttjamaican.conf.WuttaConfigExtension`

       directly.

    """

    key = 'rattail.db'

    def configure(self, config):

        """ """

        # rattail export-csv

        config.setdefault('rattail.importing', 'to_csv.from_rattail.export.default_handler',

                          'rattail.importing.exporters:FromRattailToCSV')

        config.setdefault('rattail.importing', 'to_csv.from_rattail.export.default_cmd',

                          'rattail export-csv')

        config.setdefault('rattail.importing', 'to_csv.from_rattail.export.legacy_handler_setting',

                          'rattail.exporting, csv.handler')

        # rattail export-rattail

        config.setdefault('rattail.importing', 'to_rattail.from_rattail.export.default_handler',

                          'rattail.importing.rattail:FromRattailToRattailExport')

        config.setdefault('rattail.importing', 'to_rattail.from_rattail.export.default_cmd',

                          'rattail export-rattail')

        config.setdefault('rattail.importing', 'to_rattail.from_rattail.export.legacy_handler_setting',

                          'rattail.exporting, rattail.handler')

        # rattail import-csv

        config.setdefault('rattail.importing', 'to_rattail.from_csv.import.default_handler',

                          'rattail.importing.csv:FromCSVToRattail')

        config.setdefault('rattail.importing', 'to_rattail.from_csv.import.default_cmd',

                          'rattail import-csv')

        config.setdefault('rattail.importing', 'to_rattail.from_csv.import.legacy_handler_setting',

                          'rattail.importing, csv.handler')

        # rattail import-ifps

        config.setdefault('rattail.importing', 'to_rattail.from_ifps.import.default_handler',

                          'rattail.importing.ifps:FromIFPSToRattail')

        config.setdefault('rattail.importing', 'to_rattail.from_ifps.import.default_cmd',

                          'rattail import-ifps')

        config.setdefault('rattail.importing', 'to_rattail.from_ifps.import.legacy_handler_setting',

                          'rattail.importing, ifps.handler')

        # rattail import-rattail

        config.setdefault('rattail.importing', 'to_rattail.from_rattail.import.default_handler',

                          'rattail.importing.rattail:FromRattailToRattailImport')

        config.setdefault('rattail.importing', 'to_rattail.from_rattail.import.default_cmd',

                          'rattail import-rattail')

        config.setdefault('rattail.importing', 'to_rattail.from_rattail.import.legacy_handler_setting',

                          'rattail.importing, rattail.handler')

        # rattail import-rattail-bulk

        config.setdefault('rattail.importing', 'to_rattail.from_rattail_bulk.import.default_handler',

                          'rattail.importing.rattail_bulk:BulkFromRattailToRattail')

        config.setdefault('rattail.importing', 'to_rattail.from_rattail_bulk.import.default_cmd',

                          'rattail import-rattail-bulk')

        config.setdefault('rattail.importing', 'to_rattail.from_rattail_bulk.import.legacy_handler_setting',

                          'rattail.importing, rattail_bulk.handler')

        # rattail import-sample

        config.setdefault('rattail.importing', 'to_rattail.from_sample.import.default_handler',

                          'rattail.importing.sample:FromSampleToRattail')

        config.setdefault('rattail.importing', 'to_rattail.from_sample.import.default_cmd',

                          'rattail import-sample')

        config.setdefault('rattail.importing', 'to_rattail.from_sample.import.legacy_handler_setting',

                          'rattail.importing, sample.handler')

        # rattail import-versions

        config.setdefault('rattail.importing', 'to_rattail_versions.from_rattail.import.default_handler',

                          'rattail.importing.versions:FromRattailToRattailVersions')

        config.setdefault('rattail.importing', 'to_rattail_versions.from_rattail.import.default_cmd',

                          'rattail import-versions')

        config.setdefault('rattail.importing', 'to_rattail_versions.from_rattail.import.legacy_handler_setting',

                          'rattail.importing, versions.handler')

        # trainwreck export-trainwreck

        config.setdefault('rattail.importing', 'to_trainwreck.from_trainwreck.export.default_handler',

                          'rattail.trainwreck.importing.trainwreck:FromTrainwreckToTrainwreckExport')

        config.setdefault('rattail.importing', 'to_trainwreck.from_trainwreck.export.default_cmd',

                          'trainwreck export-trainwreck')

        config.setdefault('rattail.importing', 'to_trainwreck.from_trainwreck.export.legacy_handler_setting',

                          'trainwreck.exporting, trainwreck.handler')

        # trainwreck import-self

        config.setdefault('rattail.importing', 'to_trainwreck.from_self.import.default_cmd',

                          'trainwreck import-self')

        # trainwreck import-trainwreck

        config.setdefault('rattail.importing', 'to_trainwreck.from_trainwreck.import.default_handler',

                          'rattail.trainwreck.importing.trainwreck:FromTrainwreckToTrainwreckImport')

        config.setdefault('rattail.importing', 'to_trainwreck.from_trainwreck.import.default_cmd',

                          'trainwreck import-trainwreck')

        config.setdefault('rattail.importing', 'to_trainwreck.from_trainwreck.import.legacy_handler_setting',

                          'trainwreck.importing, trainwreck.handler')

        # nb. cannot fetch DB settings during init

        appdir = config.appdir(require=False, usedb=False)

        if not appdir:

            appdir = os.path.join(sys.prefix, 'app')

        poser = config.get('rattail', 'poser', usedb=False,

                           default=os.path.join(appdir, 'poser'))

        # add poser to path if it exists

        if os.path.isdir(poser) and poser not in sys.path:

            sys.path.append(poser)

import warnings

    from .sess import Session, SessionBase as TrueSessionBase

except ImportError: # pragma: no cover

    SessionBase = None

else:

    class SessionBase(TrueSessionBase):

        def __init__(self, *args, **kwargs):

            warnings.warn("SessionBase should not be imported from rattail.db; "

                          "please import it from rattail.db.sess instead",

                          DeprecationWarning, stacklevel=2)

            super().__init__(*args, **kwargs)

# -*- coding: utf-8; -*-

################################################################################

#

#  Rattail -- Retail Software Framework

#  Copyright © 2010-2024 Lance Edgar

#

#  This file is part of Rattail.

#

#  Rattail is free software: you can redistribute it and/or modify it under the

#  terms of the GNU General Public License as published by the Free Software

#  Foundation, either version 3 of the License, or (at your option) any later

#  version.

#

#  Rattail is distributed in the hope that it will be useful, but WITHOUT ANY

#  WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

#  FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more

#  details.

#

#  You should have received a copy of the GNU General Public License along with

#  Rattail.  If not, see <http://www.gnu.org/licenses/>.

#

################################################################################

"""

Database Sessions

"""

import logging

import sqlalchemy

from sqlalchemy import orm

log = logging.getLogger(__name__)

class SessionBase(orm.Session):

    """

    Custom SQLAlchemy session base class, which adds some

    convenience methods related to the SQLAlchemy-Continuum

    integration.

    You should not instantiate this class directly; instead just

    use :class:`Session`.

    :param continuum_user: Optional user for Continuum versioning

       authorship.  If specified, the value is passed to

       :meth:`set_continuum_user()`.

    """

    def __init__(

            self,

            rattail_config=None,

            rattail_record_changes=None,

            continuum_user=None,

            **kwargs,

    ):

        super().__init__(**kwargs)

        self.rattail_config = rattail_config

        # maybe record changes

        if rattail_record_changes is None:

            rattail_record_changes = getattr(self.bind, 'rattail_record_changes', False)

        if rattail_record_changes:

            from rattail.db.changes import record_changes

            record_changes(self, config=self.rattail_config)

        else:

            self.rattail_record_changes = False

        if continuum_user is None:

            self.continuum_user = None

        else:

            self.set_continuum_user(continuum_user)

        # maybe log the current db pool status

        if getattr(self.bind, 'rattail_log_pool_status', False):

            log.debug(self.bind.pool.status())

    def set_continuum_user(self, user_info):

        """

        Set the effective Continuum user for the session.

        :param user_info: May be a

          :class:`~rattail.db.model.users.User` instance, or the

          ``uuid`` or ``username`` for one.

        """

        if self.rattail_config:

            app = self.rattail_config.get_app()

            model = app.model

        else:

            from rattail.db import model

        if isinstance(user_info, model.User):

            user = self.merge(user_info)

        else:

            user = self.get(model.User, user_info)

            if not user:

                try:

                    user = self.query(model.User).filter_by(username=user_info).one()

                except orm.exc.NoResultFound:

                    user = None

        self.continuum_user = user

Session = orm.sessionmaker(class_=SessionBase, rattail_config=None, expire_on_commit=False)

# -*- coding: utf-8; -*-

from wuttjamaican.testing import FileConfigTestCase

from rattail.config import RattailConfig

class DataTestCase(FileConfigTestCase):

    """

    Base class for test suites requiring a full (typical) database.

    """

    def setUp(self):

        self.setup_db()

    def setup_db(self):

        self.setup_files()

        self.config = self.make_config()

        self.app = self.config.get_app()

        # init db

        model = self.app.model

        model.Base.metadata.create_all(bind=self.config.appdb_engine)

        self.session = self.app.make_session()

    def tearDown(self):

        self.teardown_db()

    def teardown_db(self):

        self.teardown_files()

    def make_config(self):

        return RattailConfig(defaults={

            'rattail.db.default.url': 'sqlite://',

        })

from wuttjamaican.conf import WuttaConfigExtension

class TrainwreckConfig(WuttaConfigExtension):

# -*- coding: utf-8; -*-

from unittest import TestCase

from rattail import db as mod

class TestModule(TestCase):

    def test_basic(self):

        try:

            from sqlalchemy import orm

        except ImportError:

            pytest.skip("test is not relevant without sqlalchemy")

        # Session is legitimate

        self.assertTrue(isinstance(mod.Session, orm.sessionmaker))

        # SessionBase warns of deprecation

        # (we don't test for that, just here for the coverage)

        session = mod.SessionBase()

# -*- coding: utf-8; -*-

from unittest.mock import patch

from rattail import db

from rattail.testing import DataTestCase

try:

    import sqlalchemy as sa

    from rattail.db import sess as mod

except ImportError:

    pass

else:

    class TestSession(DataTestCase):

        def setUp(self):

            self.setup_db()

            model = self.app.model

            self.user = model.User(username='barney')

            self.session.add(self.user)

            self.session.commit()

        def test_init_rattail_config(self):

            mod.Session.configure(rattail_config=None)

            session = mod.Session()

            self.assertIsNone(session.rattail_config)

            session.close()

            config = object()

            session = mod.Session(rattail_config=config)

            self.assertIs(session.rattail_config, config)

            session.close()

        def test_init_record_changes(self):

            if hasattr(mod.Session, 'kw'):

                self.assertIsNone(mod.Session.kw.get('rattail_record_changes'))

            session = mod.Session()

            self.assertFalse(session.rattail_record_changes)

            session.close()

            session = mod.Session(rattail_record_changes=True)

            self.assertTrue(session.rattail_record_changes)

            session.close()

            engine = sa.create_engine('sqlite://')

            engine.rattail_record_changes = True

            session = mod.Session(bind=engine)

            self.assertTrue(session.rattail_record_changes)

            session.close()

        def test_continuum_user_param(self):

            # null by default

            session = mod.Session()

            self.assertIsNone(session.continuum_user)

            # can pass user

            session = mod.Session(continuum_user=self.user)

            self.assertEqual(session.continuum_user.uuid, self.user.uuid)

            # can pass username

            session = mod.Session(continuum_user='barney')

            self.assertEqual(session.continuum_user.uuid, self.user.uuid)

        def test_log_pool_status(self):

            # sanity/coverage check

            self.config.appdb_engine.rattail_log_pool_status = True

            session = mod.Session()

        def test_set_continuum_user(self):

            session = mod.Session()

            # null by default

            self.assertIsNone(session.continuum_user)

            # can pass user

            session.set_continuum_user(self.user)

            self.assertEqual(session.continuum_user.uuid, self.user.uuid)

            # can pass username

            session.set_continuum_user('barney')

            self.assertEqual(session.continuum_user.uuid, self.user.uuid)

            # null if bad username

            session.set_continuum_user('doesnotexist')

            self.assertIsNone(session.continuum_user)

            # still works if session has no config

            # TODO: not sure why session actually has a config here?

            # (guessing it is an artifact of running other tests?)

            self.assertIsNotNone(session.rattail_config)

            with patch.object(session, 'rattail_config', new=None):

                self.assertIsNone(session.rattail_config)

                session.set_continuum_user('barney')

            self.assertEqual(session.continuum_user.uuid, self.user.uuid)

    from rattail.db import model, Session

    from rattail.db.sess import SessionBase

import tempfile

            # nb. session also has our config now

            self.assertIs(session.rattail_config, config)

        else:

            # no sqlalchemy, so no rattail engines

            self.assertFalse(hasattr(config, 'rattail_engines'))

            self.assertFalse(hasattr(config, 'rattail_engine'))

            # nb. "record changes" off by default

            self.assertFalse(hasattr(db.Session, 'rattail_record_changes'))

            self.assertFalse(session.rattail_record_changes)

        # with "record changes"

        config = self.make_config(defaults={

            'rattail.db.default.url': 'sqlite://',

            'rattail.db.changes.record': 'true',

        })

        if db.Session:

            self.assertTrue(db.Session.rattail_record_changes)

            session = db.Session()

            # nb. session instance still does not have the flag

            # TODO: is this by design..?  does it need to improve?

            self.assertFalse(session.rattail_record_changes)

class TestLegacyConfigExtensionBase(TestCase):

    def test_basic(self):

        # sanity / coverage check

        ext = mod.ConfigExtension()

class TestRattailConfigExtension(TestCase):

    def make_config(self, **kwargs):

        return mod.RattailConfig(**kwargs)

    def make_extension(self):

        return mod.RattailConfigExtension()

    def test_configure(self):

        # no import config yet

        config = self.make_config()

        self.assertEqual(config.defaults, {})

        self.assertIsNone(config.get('rattail.importing.to_rattail.from_csv.import.default_handler'))

        # extension adds import config

        ext = self.make_extension()

        ext.configure(config)