opendev/storyboard
Code Issues Proposed changes

Merge "Put the logic for hiding private things in storyboard/db/api/base.py"

Browse Source
This commit is contained in:
Jenkins 2016-08-24 17:11:17 +00:00 committed by Gerrit Code Review
commit 5bafba1140
7 changed files with 183 additions and 259 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
storyboard/api/v1/search/sqlalchemy_impl.py
View File

@ -13,9 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from sqlalchemy import and_, or_
from sqlalchemy.orm import subqueryload
from sqlalchemy.sql.expression import false, true
from sqlalchemy_fulltext import FullTextSearch
import sqlalchemy_fulltext.modes as FullTextMode
@ -62,22 +60,7 @@ class SqlAlchemySearchImpl(search_engine.SearchEngine):
subquery = api_base.model_query(models.Story, session)
# Filter out stories that the current user can't see
subquery = subquery.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
subquery = subquery.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
subquery = subquery.filter(models.Story.private == false())
subquery = api_base.filter_private_stories(subquery, current_user)
subquery = self._build_fulltext_search(models.Story, subquery, q)
subquery = self._apply_pagination(models.Story,
@ -99,23 +82,8 @@ class SqlAlchemySearchImpl(search_engine.SearchEngine):
query = api_base.model_query(models.Task, session)
# Filter out tasks or stories that the current user can't see
query = query.outerjoin(models.Story,
models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
query = query.filter(models.Story.private == false())
query = query.outerjoin(models.Story)
query = api_base.filter_private_stories(query, current_user)
query = self._build_fulltext_search(models.Task, query, q)
query = self._apply_pagination(

163
storyboard/db/api/base.py
View File

@ -23,6 +23,9 @@ from oslo_db.sqlalchemy.utils import paginate_query as utils_paginate_query
from oslo_log import log
from pecan import request
import six
from sqlalchemy import and_, or_
from sqlalchemy.orm import aliased
from sqlalchemy.sql.expression import false, true
import sqlalchemy.types as sqltypes
from storyboard.common import exception as exc
@ -371,3 +374,163 @@ def entity_hard_delete(kls, entity_id, session=None):
raise exc.DBDeadLock()
except db_exc.DBInvalidUnicodeParameter:
raise exc.DBInvalidUnicodeParameter()
def filter_private_stories(query, current_user, story_model=models.Story):
"""Takes a query and filters out stories the user shouldn't see.
:param query: The query to be filtered.
:param current_user: The ID of the user requesting the result.
:param story_model: The database model used for stories in the query.
"""
query = query.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user:
query = query.filter(
or_(
and_(
models.User.id == current_user,
story_model.private == true()
),
story_model.private == false(),
story_model.id.is_(None)
)
)
else:
query = query.filter(
or_(
story_model.private == false(),
story_model.id.is_(None)
)
)
return query
def filter_private_worklists(query, current_user, hide_lanes=True):
"""Takes a query and filters out worklists the user shouldn't see.
:param query: The query to be filtered.
:param current_user: The ID of the user requesting the result.
:param hide_lanes: Whether or not to also filter out worklists which
are lanes in a board.
"""
# Alias all the things we're joining, in case they've been
# joined already.
board_worklists = aliased(models.BoardWorklist)
boards = aliased(models.Board)
board_permissions = aliased(models.board_permissions)
worklist_permissions = aliased(models.worklist_permissions)
permissions = aliased(models.Permission)
user_permissions = aliased(models.user_permissions)
users = aliased(models.User)
# Worklists permissions must be inherited from the board which
# contains the list (if any). To handle this we split the query
# into the lists which are in boards (`lanes`) and those which
# aren't (`lists`). We then either hide the lanes entirely or
# unify the two queries.
lanes = query.outerjoin(
(board_worklists, models.Worklist.id == board_worklists.list_id))
lanes = (lanes
.outerjoin((boards, boards.id == board_worklists.board_id))
.outerjoin((board_permissions,
boards.id == board_permissions.c.board_id))
.outerjoin((permissions,
permissions.id == board_permissions.c.permission_id))
.outerjoin((user_permissions,
permissions.id == user_permissions.c.permission_id))
.outerjoin((users, user_permissions.c.user_id == users.id)))
lists = query.outerjoin(
(board_worklists, models.Worklist.id == board_worklists.list_id))
lists = (lists.filter(board_worklists.board_id.is_(None))
.outerjoin((worklist_permissions,
models.Worklist.id == worklist_permissions.c.worklist_id))
.outerjoin((permissions,
permissions.id == worklist_permissions.c.permission_id))
.outerjoin((user_permissions,
user_permissions.c.permission_id == permissions.id))
.outerjoin((users, user_permissions.c.user_id == users.id)))
if current_user:
if not hide_lanes:
lanes = lanes.filter(
or_(
and_(
users.id == current_user,
boards.private == true()
),
boards.private == false()
)
)
lists = lists.filter(
or_(
and_(
users.id == current_user,
boards.private == true()
),
models.Worklist.private == false(),
models.Worklist.id.is_(None)
)
)
else:
if not hide_lanes:
lanes = lanes.filter(boards.private == false())
lists = lists.filter(
or_(
models.Worklist.private == false(),
models.Worklist.id.is_(None)
)
)
if hide_lanes:
return lists
return lists.union(lanes)
def filter_private_boards(query, current_user):
"""Takes a query and filters out the boards that the user should not see.
:param query: The query to be filtered.
:param current_user: The ID of the user requesting the result.
"""
board_permissions = aliased(models.board_permissions)
permissions = aliased(models.Permission)
user_permissions = aliased(models.user_permissions)
users = aliased(models.User)
query = (query
.outerjoin((board_permissions,
models.Board.id == board_permissions.c.board_id))
.outerjoin((permissions,
board_permissions.c.permission_id == permissions.id))
.outerjoin((user_permissions,
permissions.id == user_permissions.c.permission_id))
.outerjoin((users, user_permissions.c.user_id == users.id)))
if current_user:
query = query.filter(
or_(
and_(
users.id == current_user,
models.Board.private == true()
),
models.Board.private == false(),
models.Board.id.is_(None)
)
)
else:
query = query.filter(
or_(
models.Board.private == false(),
models.Board.id.is_(None)
)
)
return query

19
storyboard/db/api/boards.py
View File

@ -13,9 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from sqlalchemy import and_, or_
from sqlalchemy.orm import aliased, subqueryload
from sqlalchemy.sql.expression import false, true
from wsme.exc import ClientSideError
from storyboard.db.api import base as api_base
@ -50,22 +48,7 @@ def _build_board_query(title=None, creator_id=None, user_id=None,
project_id=project_id)
# Filter out boards that the current user can't see
query = query.join(models.board_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.Board.private == true()
),
models.Board.private == false()
)
)
else:
query = query.filter(models.Board.private == false())
query = api_base.filter_private_boards(query, current_user)
# Filter by boards that a given user has permissions to use
if user_id:

41
storyboard/db/api/due_dates.py
View File

@ -13,8 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from sqlalchemy import and_, func, or_
from sqlalchemy.sql.expression import false, true
from sqlalchemy import func
from wsme.exc import ClientSideError
from storyboard.db.api import base as api_base
@ -111,41 +110,9 @@ def update(id, values):
def get_visible_items(due_date, current_user=None):
stories = due_date.stories.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
due_date.stories = due_date.stories.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
due_date.stories = due_date.stories.filter(
models.Story.private == false())
tasks = due_date.tasks.outerjoin(models.Story,
models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
tasks = tasks.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
tasks = tasks.filter(models.Story.private == false())
stories = api_base.filter_private_stories(due_date.stories, current_user)
tasks = due_date.tasks.outerjoin(models.Story)
tasks = api_base.filter_private_stories(tasks, current_user)
return stories, tasks

54
storyboard/db/api/stories.py
View File

@ -13,9 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from sqlalchemy import and_, or_
from sqlalchemy.orm import subqueryload
from sqlalchemy.sql.expression import false, true
from storyboard.common import exception as exc
from storyboard.db.api import base as api_base
@ -32,22 +30,7 @@ def story_get_simple(story_id, session=None, current_user=None):
.filter_by(id=story_id)
# Filter out stories that the current user can't see
query = query.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
query = query.filter(models.Story.private == false())
query = api_base.filter_private_stories(query, current_user)
return query.first()
@ -58,22 +41,8 @@ def story_get(story_id, session=None, current_user=None):
.filter_by(id=story_id)
# Filter out stories that the current user can't see
query = query.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.StorySummary.private == true()
),
models.StorySummary.private == false()
)
)
else:
query = query.filter(models.StorySummary.private == false())
query = api_base.filter_private_stories(query, current_user,
story_model=models.StorySummary)
return query.first()
@ -188,22 +157,7 @@ def _story_build_query(title=None, description=None, assignee_id=None,
creator_id=creator_id)
# Filter out stories that the current user can't see
query = query.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
query = query.filter(models.Story.private == false())
query = api_base.filter_private_stories(query, current_user)
# Filtering by tags
if tags:

41
storyboard/db/api/tasks.py
View File

@ -13,9 +13,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from sqlalchemy import and_, or_
from sqlalchemy.sql.expression import false, true
from storyboard.db.api import base as api_base
from storyboard.db import models
@ -25,23 +22,8 @@ def task_get(task_id, session=None, current_user=None):
query = query.filter(models.Task.id == task_id)
# Filter out tasks or stories that the current user can't see
query = query.outerjoin(models.Story,
models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
query = query.filter(models.Story.private == false())
query = query.outerjoin(models.Story)
query = api_base.filter_private_stories(query, current_user)
return query.first()
@ -106,23 +88,8 @@ def task_build_query(project_group_id, current_user=None, **kwargs):
**kwargs)
# Filter out tasks or stories that the current user can't see
query = query.outerjoin(models.Story,
models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
query = query.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
query = query.filter(models.Story.private == false())
query = query.outerjoin(models.Story)
query = api_base.filter_private_stories(query, current_user)
return query

86
storyboard/db/api/worklists.py
View File

@ -13,9 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from sqlalchemy import and_, or_
from sqlalchemy.orm import aliased
from sqlalchemy.sql.expression import false, true
from wsme.exc import ClientSideError
from storyboard.common import exception as exc
@ -51,53 +49,7 @@ def _build_worklist_query(title=None, creator_id=None, project_id=None,
creator_id=creator_id,
project_id=project_id)
# Filter out lists that the current user can't see.
# This gets complicated because worklists permissions must be
# inherited from the board which contains the list (if any). To
# handle this we split the query into the lists which are in
# boards (`lanes`) and those which aren't (`lists`). We then
# either hide the lanes entirely or unify the two queries.
lanes = query.join(models.BoardWorklist,
models.Board,
models.board_permissions)
lanes = lanes.join(models.Permission,
models.user_permissions,
models.User)
lists = query.outerjoin(models.BoardWorklist)
lists = lists.filter(models.BoardWorklist.board_id.is_(None))
lists = lists.join(models.worklist_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user:
if not hide_lanes:
lanes = lanes.filter(
or_(
and_(
models.User.id == current_user,
models.Board.private == true()
),
models.Board.private == false()
)
)
lists = lists.filter(
or_(
and_(
models.User.id == current_user,
models.Worklist.private == true()
),
models.Worklist.private == false()
)
)
else:
if not hide_lanes:
lanes = lanes.filter(models.Board.private == false())
lists = lists.filter(models.Worklist.private == false())
if hide_lanes:
query = lists
else:
query = lists.union(lists)
query = api_base.filter_private_worklists(query, current_user)
# Filter by lists that a given user has permissions to use
if user_id:
@ -231,43 +183,13 @@ def get_visible_items(worklist, current_user=None):
stories = worklist.items.filter(models.WorklistItem.item_type == 'story')
stories = stories.join(
(models.Story, models.Story.id == models.WorklistItem.item_id))
stories = stories.outerjoin(models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
stories = stories.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
stories = stories.filter(models.Story.private == false())
stories = api_base.filter_private_stories(stories, current_user)
tasks = worklist.items.filter(models.WorklistItem.item_type == 'task')
tasks = tasks.join(
(models.Task, models.Task.id == models.WorklistItem.item_id))
tasks = tasks.outerjoin(models.Story,
models.story_permissions,
models.Permission,
models.user_permissions,
models.User)
if current_user is not None:
tasks = tasks.filter(
or_(
and_(
models.User.id == current_user,
models.Story.private == true()
),
models.Story.private == false()
)
)
else:
tasks = tasks.filter(models.Story.private == false())
tasks = tasks.outerjoin(models.Story)
tasks = api_base.filter_private_stories(tasks, current_user)
return stories.union(tasks)