-
Notifications
You must be signed in to change notification settings - Fork 1
/
base.py
222 lines (187 loc) · 6.5 KB
/
base.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
from abc import ABC, abstractmethod
from enum import IntFlag, auto
from functools import wraps
from inspect import signature
from typing import Any, Optional, Dict, Tuple
from typing import Mapping
from starlette.exceptions import HTTPException
from starlette.status import (
HTTP_404_NOT_FOUND,
HTTP_409_CONFLICT,
HTTP_403_FORBIDDEN,
HTTP_400_BAD_REQUEST,
)
from ....database import Database
from ....logging import log
from ....security import User
class Status(IntFlag):
NONE = auto()
EXISTS = auto()
DOES_NOT_EXIST = auto()
PUBLISHED = auto()
NOT_PUBLISHED = auto()
OWN = auto()
ANONYMOUS = auto()
AUTHENTICATED = auto()
ADMIN = auto()
SUPERUSER = auto()
ADMIN_POSTING = auto()
AUTO_PUBLISH = auto()
@staticmethod
def construct(
status_data: Optional[Mapping[str, Any]],
status_mapping: Dict[str, Tuple["Status", "Status"]],
) -> "Status":
status = Status.NONE
if status_data is not None:
for key, (on_true, on_false) in status_mapping.items():
if key in status_data and status_data[key]:
status |= on_true
else:
status |= on_false
else:
status |= status.DOES_NOT_EXIST
return status
class StatusProvider(ABC):
"""Checks resource existence status and any prerequisites"""
db: Database
user: User
identifiers: Dict[str, Any]
async def provide_status(self) -> Status:
status = Status.NONE
if self.user.is_authenticated:
status |= Status.AUTHENTICATED
if self.user.is_superuser:
status |= Status.SUPERUSER | Status.ADMIN
else:
status |= Status.ANONYMOUS
return status | await self.derive_status(await self.query())
async def query(self):
if self.user.is_authenticated:
return await self.db.fetch_one(
self.query_authenticated,
{
"user": self.user.identity,
**self.identifiers,
},
)
else:
return await self.db.fetch_one(
self.query_anonymous,
{
**self.identifiers,
},
)
@property
@abstractmethod
def query_authenticated(self) -> str:
"""Query when the user is authenticated"""
@property
@abstractmethod
def query_anonymous(self) -> str:
"""Query when the user is not authenticated"""
@abstractmethod
async def derive_status(self, m: Mapping) -> Status:
"""Checks the existence status for this instance
This should first check that all prerequisites are filled and throw if not
and then use the methods from Status to construct a Status.
Mainly useful for the state decorators.
Returns
-------
status
Status-instance to check the state of this resource
Raises
------
fastapi.HTTPException
On failure to meet prerequisites
"""
def require_status(
*acceptable_statuses: Status, errors: Dict[Status, HTTPException] = None
):
errors = {} if not errors else {**errors}
privilege_error = HTTPException(
status_code=HTTP_403_FORBIDDEN,
detail="Not enough privileges",
)
not_found_error = HTTPException(
status_code=HTTP_404_NOT_FOUND,
detail="Resource not found",
)
default_exception_mappers = {
Status.DOES_NOT_EXIST: not_found_error,
Status.NOT_PUBLISHED: not_found_error,
Status.ANONYMOUS: privilege_error,
Status.EXISTS: HTTPException(
status_code=HTTP_409_CONFLICT,
detail="Resource already exists",
),
}
authenticated_statuses = [
Status.OWN,
Status.AUTHENTICATED,
Status.ADMIN,
Status.SUPERUSER,
]
if any(
status in s for status in authenticated_statuses for s in acceptable_statuses
):
errors[Status.EXISTS | Status.PUBLISHED | Status.ANONYMOUS] = privilege_error
if any(Status.ADMIN in s for s in acceptable_statuses):
errors[
Status.EXISTS | Status.PUBLISHED | Status.AUTHENTICATED
] = privilege_error
if any(Status.SUPERUSER in s or Status.OWN in s for s in acceptable_statuses):
errors[Status.EXISTS | Status.PUBLISHED | Status.ANONYMOUS] = privilege_error
errors[
Status.EXISTS | Status.PUBLISHED | Status.AUTHENTICATED
] = privilege_error
errors[
Status.EXISTS | Status.PUBLISHED | Status.AUTHENTICATED
] = privilege_error
errors[Status.EXISTS | Status.SUPERUSER] = privilege_error
errors[Status.EXISTS | Status.ADMIN] = privilege_error
def decorator(f):
status_parameters = [
name
for name, param in signature(f).parameters.items()
if param.annotation == Status
]
function_name = f.__name__
@wraps(f)
async def requirement(self: StatusProvider, *args, **kwargs):
status = await self.provide_status()
if any(option & status == option for option in acceptable_statuses):
if status_parameters:
return await f(
self,
*args,
**kwargs,
**{name: status for name in status_parameters}
)
else:
return await f(self, *args, **kwargs)
else:
log.error(
"Attempted to call %s() with invalid status:"
"\n -expected: %r"
"\n -presented: %r"
"\n -context: %r"
"\n -user: %r",
function_name,
acceptable_statuses,
status,
self.identifiers,
self.user.identity if self.user.is_authenticated else "",
)
for error_status, error in (
*errors.items(),
*default_exception_mappers.items(),
):
if error_status & status == error_status:
raise error
raise HTTPException(
status_code=HTTP_400_BAD_REQUEST,
detail="Request failed to meet required status",
)
return requirement
return decorator