Raise custom error if unbound session

ftrack_query/__init__.py

@@ -7,16 +7,16 @@
 orientated approach. Inspiration was taken from SQLALchemy.
 """
 
-__all__ = ['FTrackQuery', 'entity', 'and_', 'or_', 'not_', 'event',
+__all__ = ['FTrackQuery', 'exception', 'entity', 'and_', 'or_', 'not_', 'event',
            'select', 'create', 'update', 'delete', 'attr']
 
-__version__ = '1.8.2'
+__version__ = '1.8.3'
 
 import os
 
 import ftrack_api
 
-from . import utils
+from . import exception, utils
 from .abstract import AbstractStatement
 from .query import Query, entity, and_, or_, not_
 from .event import event

ftrack_query/exception.py

@@ -0,0 +1,6 @@
+class UnboundSessionError(Exception):
+    """Raise an error if attempting to execute a statement with no session."""
+
+    def __init__(self):
+        msg = 'statement has no session bound to it'
+        super(UnboundSessionError, self).__init__(msg)

ftrack_query/query.py

@@ -25,6 +25,7 @@
 from ftrack_api.symbol import NOT_SET
 
 from .abstract import AbstractComparison, AbstractQuery, AbstractStatement
+from .exception import UnboundSessionError
 from .utils import Join, clone_instance, convert_output_value, not_, parse_operators
 
 
@@ -335,6 +336,9 @@ def get(self, value, _value=None):
         """Get an entity from the ID.
         The _value argument is for compatibility with ftrack_api.Session.
         """
+        if self._session is None:
+            raise UnboundSessionError
+
         if _value is None:
             entity = self._entity
         else:
@@ -343,16 +347,22 @@ def get(self, value, _value=None):
 
     def create(self, **kwargs):
         """Create a new entity."""
+        if self._session is None:
+            raise UnboundSessionError
         return self._session.create(self._entity, kwargs)
 
     def ensure(self, **kwargs):
         """Ensure an entity.
         Will create if it doesn't exist.
         """
+        if self._session is None:
+            raise UnboundSessionError
         return self._session.ensure(self._entity, kwargs)
 
     def _exec_query(self):
         """Execute the current query."""
+        if self._session is None:
+            raise UnboundSessionError
         return self._session.query(self.as_str(), page_size=self._page_size)
 
     def one(self):

ftrack_query/statement.py

@@ -21,6 +21,7 @@
 from ftrack_api.symbol import NOT_SET
 
 from .abstract import AbstractStatement
+from .exception import UnboundSessionError
 from .query import Query, Comparison
 from .utils import clone_instance, copy_doc
 
@@ -110,6 +111,8 @@ def execute(self, session=None):
         This does not commit changes.
         """
         if session is None:
+            if self._session is None:
+                raise UnboundSessionError
             session = self._session
         return session.create(self._entity, self._values)
 
@@ -146,6 +149,8 @@ def execute(self, session=None):
         This does not commit changes.
         """
         if session is None:
+            if self._session is None:
+                raise UnboundSessionError
             session = self._session
 
         count = 0
@@ -210,6 +215,8 @@ def copy(self):
     def execute(self, session=None):
         """Execute the select statement."""
         if session is None:
+            if self._session is None:
+                raise UnboundSessionError
             session = self._session
 
         count = 0

tests/test_v2.py

@@ -3,7 +3,7 @@
 import sys
 
 sys.path.insert(0, os.path.normpath(os.path.dirname(__file__)).rsplit(os.path.sep, 1)[0])
-from ftrack_query import FTrackQuery, attr, entity, select
+from ftrack_query import FTrackQuery, exception, attr, entity, select
 
 
 class TestSelect(unittest.TestCase):
@@ -58,5 +58,17 @@ def test_page_size(self):
         self.assertEqual(query._page_size, 50)
 
 
+class TestException(unittest.TestCase):
+    def setUp(self):
+        self.session = FTrackQuery(debug=True)
+
+    def testUnboundSession(self):
+        with self.assertRaises(exception.UnboundSessionError):
+            select('Task').first()
+
+        with self.assertRaises(exception.UnboundSessionError):
+            self.session.select('Task').options(session=None).first()
+
+
 if __name__ == '__main__':
     unittest.main()