Factor out queries

atst/domain/workspaces/query.py

@@ -0,0 +1,64 @@
+from sqlalchemy.orm.exc import NoResultFound
+
+from atst.database import db
+from atst.domain.exceptions import NotFoundError
+from atst.models.workspace import Workspace
+from atst.models.workspace_role import WorkspaceRole
+
+
+class Query(object):
+
+    model = None
+
+    @property
+    def resource_name(cls):
+        return cls.model.__class__.lower()
+
+    @classmethod
+    def create(cls, **kwargs):
+        # pylint: disable=E1102
+        return cls.model(**kwargs)
+
+    @classmethod
+    def get(cls, id_):
+        try:
+            resource = db.session.query(cls.model).filter_by(id=id_).one()
+            return resource
+        except NoResultFound:
+            raise NotFoundError(cls.resource_name)
+
+    @classmethod
+    def get_all(cls):
+        return db.session.query(cls.model).all()
+
+    @classmethod
+    def add_and_commit(cls, resource):
+        db.session.add(resource)
+        db.session.commit()
+        return resource
+
+
+class WorkspaceQuery(Query):
+    model = Workspace
+
+    @classmethod
+    def get_by_request(cls, request):
+        try:
+            workspace = db.session.query(Workspace).filter_by(request=request).one()
+        except NoResultFound:
+            raise NotFoundError("workspace")
+
+        return workspace
+
+    @classmethod
+    def get_for_user(cls, user):
+        return (
+            db.session.query(Workspace)
+            .join(WorkspaceRole)
+            .filter(WorkspaceRole.user == user)
+            .all()
+        )
+
+    @classmethod
+    def create_workspace_role(cls, user, role, workspace):
+        return WorkspaceRole(user=user, role=role, workspace=workspace)

atst/domain/workspaces/workspaces.py

@@ -1,14 +1,10 @@
-from sqlalchemy.orm.exc import NoResultFound
-
-from atst.database import db
-from atst.models.workspace import Workspace
-from atst.models.workspace_role import WorkspaceRole
-from atst.domain.exceptions import NotFoundError
 from atst.domain.roles import Roles
 from atst.domain.authz import Authorization
 from atst.models.permissions import Permissions
 from atst.domain.users import Users
 from atst.domain.workspace_users import WorkspaceUsers
+
+from .query import WorkspaceQuery
 from .scopes import ScopedWorkspace
 
 
@@ -16,17 +12,14 @@ class Workspaces(object):
     @classmethod
     def create(cls, request, name=None):
         name = name or request.id
-        workspace = Workspace(request=request, name=name)
+        workspace = WorkspaceQuery.create(request=request, name=name)
         Workspaces._create_workspace_role(request.creator, workspace, "owner")
-
+        WorkspaceQuery.add_and_commit(workspace)
-        db.session.add(workspace)
-        db.session.commit()
-
         return workspace
 
     @classmethod
     def get(cls, user, workspace_id):
-        workspace = Workspaces._get(workspace_id)
+        workspace = WorkspaceQuery.get(workspace_id)
         Authorization.check_workspace_permission(
             user, workspace, Permissions.VIEW_WORKSPACE, "get workspace"
         )
@@ -35,7 +28,7 @@ class Workspaces(object):
 
     @classmethod
     def get_for_update(cls, user, workspace_id):
-        workspace = Workspaces._get(workspace_id)
+        workspace = WorkspaceQuery.get(workspace_id)
         Authorization.check_workspace_permission(
             user, workspace, Permissions.ADD_APPLICATION_IN_WORKSPACE, "add project"
         )
@@ -44,16 +37,11 @@ class Workspaces(object):
 
     @classmethod
     def get_by_request(cls, request):
-        try:
+        return WorkspaceQuery.get_by_request(request)
-            workspace = db.session.query(Workspace).filter_by(request=request).one()
-        except NoResultFound:
-            raise NotFoundError("workspace")
-
-        return workspace
 
     @classmethod
     def get_with_members(cls, user, workspace_id):
-        workspace = Workspaces._get(workspace_id)
+        workspace = WorkspaceQuery.get(workspace_id)
         Authorization.check_workspace_permission(
             user,
             workspace,
@@ -63,27 +51,12 @@ class Workspaces(object):
 
         return workspace
 
-    @classmethod
-    def get_many(cls, user):
-        workspaces = (
-            db.session.query(Workspace)
-            .join(WorkspaceRole)
-            .filter(WorkspaceRole.user == user)
-            .all()
-        )
-        return workspaces
-
     @classmethod
     def for_user(cls, user):
         if Authorization.has_atat_permission(user, Permissions.VIEW_WORKSPACE):
-            workspaces = db.session.query(Workspace).all()
+            workspaces = WorkspaceQuery.get_all()
         else:
-            workspaces = (
+            workspaces = WorkspaceQuery.get_for_user(user)
-                db.session.query(Workspace)
-                .join(WorkspaceRole)
-                .filter(WorkspaceRole.user == user)
-                .all()
-            )
         return workspaces
 
     @classmethod
@@ -122,15 +95,6 @@ class Workspaces(object):
     @classmethod
     def _create_workspace_role(cls, user, workspace, role_name):
         role = Roles.get(role_name)
-        workspace_role = WorkspaceRole(user=user, role=role, workspace=workspace)
+        workspace_role = WorkspaceQuery.create_workspace_role(user, role, workspace)
-        db.session.add(workspace_role)
+        WorkspaceQuery.add_and_commit(workspace_role)
         return workspace_role
-
-    @classmethod
-    def _get(cls, workspace_id):
-        try:
-            workspace = db.session.query(Workspace).filter_by(id=workspace_id).one()
-        except NoResultFound:
-            raise NotFoundError("workspace")
-
-        return workspace

tests/domain/test_workspaces.py

@@ -63,18 +63,6 @@ def test_workspaces_get_ensures_user_is_in_workspace(workspace, workspace_owner)
         Workspaces.get(outside_user, workspace.id)
 
 
-def test_workspaces_get_many_with_no_workspaces():
-    workspaces = Workspaces.get_many(UserFactory.build())
-    assert workspaces == []
-
-
-def test_workspaces_get_many_returns_a_users_workspaces(workspace, workspace_owner):
-    # random workspace
-    Workspaces.create(RequestFactory.create())
-
-    assert Workspaces.get_many(workspace_owner) == [workspace]
-
-
 def test_get_for_update_allows_owner(workspace, workspace_owner):
     Workspaces.get_for_update(workspace_owner, workspace.id)