Coverage for cc_modules/cc_taskfactory.py : 32%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
3"""
4camcops_server/cc_modules/cc_taskfactory.py
6===============================================================================
8 Copyright (C) 2012-2020 Rudolf Cardinal (rudolf@pobox.com).
10 This file is part of CamCOPS.
12 CamCOPS is free software: you can redistribute it and/or modify
13 it under the terms of the GNU General Public License as published by
14 the Free Software Foundation, either version 3 of the License, or
15 (at your option) any later version.
17 CamCOPS is distributed in the hope that it will be useful,
18 but WITHOUT ANY WARRANTY; without even the implied warranty of
19 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20 GNU General Public License for more details.
22 You should have received a copy of the GNU General Public License
23 along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.
25===============================================================================
27**Functions to fetch tasks from the database.**
29"""
31import logging
32from typing import Optional, Type, TYPE_CHECKING, Union
34from cardinal_pythonlib.logs import BraceStyleAdapter
35import pyramid.httpexceptions as exc
36from sqlalchemy.orm import Query, Session as SqlASession
38from camcops_server.cc_modules.cc_task import (
39 tablename_to_task_class_dict,
40 Task,
41)
42from camcops_server.cc_modules.cc_taskindex import TaskIndexEntry
44if TYPE_CHECKING:
45 from camcops_server.cc_modules.cc_request import CamcopsRequest
47log = BraceStyleAdapter(logging.getLogger(__name__))
50# =============================================================================
51# Task query helpers
52# =============================================================================
54def task_query_restricted_to_permitted_users(
55 req: "CamcopsRequest",
56 q: Query,
57 cls: Union[Type[Task], Type[TaskIndexEntry]],
58 as_dump: bool) -> Optional[Query]:
59 """
60 Restricts an SQLAlchemy ORM query to permitted users, for a given
61 task class. THIS IS A KEY SECURITY FUNCTION.
63 Args:
64 req:
65 the :class:`camcops_server.cc_modules.cc_request.CamcopsRequest`
66 q:
67 the SQLAlchemy ORM query
68 cls:
69 the class of the task type, or the
70 :class:`camcops_server.cc_modules.cc_taskindex.TaskIndexEntry`
71 class
72 as_dump:
73 use the "dump" permissions rather than the "view" permissions?
75 Returns:
76 a filtered query (or the original query, if no filtering was required)
78 """
79 user = req.user
81 if user.superuser:
82 return q # anything goes
84 # Implement group security. Simple:
85 if as_dump:
86 group_ids = user.ids_of_groups_user_may_dump
87 else:
88 group_ids = user.ids_of_groups_user_may_see
90 if not group_ids:
91 return None
93 if cls is TaskIndexEntry:
94 # noinspection PyUnresolvedReferences
95 q = q.filter(cls.group_id.in_(group_ids))
96 else: # a kind of Task
97 q = q.filter(cls._group_id.in_(group_ids))
99 return q
102# =============================================================================
103# Make a single task given its base table name and server PK
104# =============================================================================
106def task_factory(req: "CamcopsRequest", basetable: str,
107 serverpk: int) -> Optional[Task]:
108 """
109 Load a task from the database and return it.
110 Filters to tasks permitted to the current user.
112 Args:
113 req: the :class:`camcops_server.cc_modules.cc_request.CamcopsRequest`
114 basetable: name of the task's base table
115 serverpk: server PK of the task
117 Returns:
118 the task, or ``None`` if the PK doesn't exist
120 Raises:
121 :exc:`HTTPBadRequest` if the table doesn't exist
123 """
124 d = tablename_to_task_class_dict()
125 try:
126 cls = d[basetable] # may raise KeyError
127 except KeyError:
128 raise exc.HTTPBadRequest(f"No such task table: {basetable!r}")
129 dbsession = req.dbsession
130 # noinspection PyProtectedMember
131 q = dbsession.query(cls).filter(cls._pk == serverpk)
132 q = task_query_restricted_to_permitted_users(req, q, cls, as_dump=False)
133 return q.first()
136def task_factory_no_security_checks(dbsession: SqlASession, basetable: str,
137 serverpk: int) -> Optional[Task]:
138 """
139 Load a task from the database and return it.
140 Filters to tasks permitted to the current user.
142 Args:
143 dbsession: a :class:`sqlalchemy.orm.session.Session`
144 basetable: name of the task's base table
145 serverpk: server PK of the task
147 Returns:
148 the task, or ``None`` if the PK doesn't exist
150 Raises:
151 :exc:`KeyError` if the table doesn't exist
152 """
153 d = tablename_to_task_class_dict()
154 cls = d[basetable] # may raise KeyError
155 # noinspection PyProtectedMember
156 q = dbsession.query(cls).filter(cls._pk == serverpk)
157 return q.first()
160# =============================================================================
161# Make a single task given its base table name and server PK
162# =============================================================================
164def task_factory_clientkeys_no_security_checks(dbsession: SqlASession,
165 basetable: str,
166 client_id: int,
167 device_id: int,
168 era: str) -> Optional[Task]:
169 """
170 Load a task from the database and return it.
171 Filters to tasks permitted to the current user.
173 Args:
174 dbsession: a :class:`sqlalchemy.orm.session.Session`
175 basetable: name of the task's base table
176 client_id: task's ``_id`` value
177 device_id: task's ``_device_id`` value
178 era: task's ``_era`` value
180 Returns:
181 the task, or ``None`` if it doesn't exist
183 Raises:
184 :exc:`KeyError` if the table doesn't exist
185 """
186 d = tablename_to_task_class_dict()
187 cls = d[basetable] # may raise KeyError
188 # noinspection PyProtectedMember
189 q = (
190 dbsession.query(cls)
191 .filter(cls.id == client_id)
192 .filter(cls._device_id == device_id)
193 .filter(cls._era == era)
194 )
195 return q.first()