-
Notifications
You must be signed in to change notification settings - Fork 1.9k
/
user_favourite_events.py
119 lines (100 loc) · 4.01 KB
/
user_favourite_events.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
from flask import request, current_app as app
from flask_jwt_extended import current_user, jwt_required, verify_jwt_in_request
from flask_rest_jsonapi import ResourceDetail, ResourceList, ResourceRelationship
from flask_rest_jsonapi.exceptions import ObjectNotFound
from sqlalchemy.orm.exc import NoResultFound
from app.api.helpers.db import safe_query
from app.api.helpers.exceptions import ForbiddenException, ConflictException
from app.api.helpers.permission_manager import has_access
from app.api.helpers.utilities import require_relationship
from app.api.schema.user_favourite_events import UserFavouriteEventSchema
from app.models import db
from app.models.user import User
from app.models.user_favourite_event import UserFavouriteEvent
class UserFavouriteEventListPost(ResourceList):
"""
Create User Favourite Events
"""
@classmethod
def before_post(self, args, kwargs, data):
"""
before post method to check for required relationship and proper permission
:param args:
:param kwargs:
:param data:
:return:
"""
require_relationship(['event'], data)
if 'Authorization' in request.headers:
verify_jwt_in_request()
else:
raise ForbiddenException({'source': ''}, 'Only Authorized Users can favourite an event')
data['user'] = current_user.id
user_favourite_event = find_user_favourite_event_by_id(event_id=data['event'])
if user_favourite_event:
raise ConflictException({'pointer': '/data/relationships/event'}, "Event already favourited")
view_kwargs = True
schema = UserFavouriteEventSchema
methods = ['POST', ]
data_layer = {'session': db.session,
'model': UserFavouriteEvent,
'methods': {'before_post': before_post}}
class UserFavouriteEventList(ResourceList):
"""
List User Favourite Events
"""
def query(self, view_kwargs):
"""
query method for SessionList class
:param view_kwargs:
:return:
"""
query_ = self.session.query(UserFavouriteEvent)
if view_kwargs.get('user_id') is not None:
user = safe_query(self, User, 'id', view_kwargs['user_id'], 'user_id')
query_ = query_.join(User).filter(User.id == user.id)
elif has_access('is_admin'):
pass
return query_
methods = ['GET']
schema = UserFavouriteEventSchema
data_layer = {'session': db.session,
'model': UserFavouriteEvent,
'methods': {
'query': query
}}
class UserFavouriteEventDetail(ResourceDetail):
"""
User Favourite Events detail by id
"""
@jwt_required
def before_get_object(self, view_kwargs):
if view_kwargs.get('id') is not None:
try:
user_favourite_event = find_user_favourite_event_by_id(event_id=view_kwargs['id'])
except NoResultFound:
raise ObjectNotFound({'source': '/data/relationships/event'}, "Object: not found")
else:
if user_favourite_event is not None:
view_kwargs['id'] = user_favourite_event.id
else:
view_kwargs['id'] = None
methods = ['GET', 'DELETE']
schema = UserFavouriteEventSchema
data_layer = {'session': db.session,
'model': UserFavouriteEvent,
'methods': {
'before_get_object': before_get_object,
}}
class UserFavouriteEventRelationship(ResourceRelationship):
"""
User Favourite Events Relationship
"""
schema = UserFavouriteEventSchema
methods = ['GET']
data_layer = {'session': db.session,
'model': UserFavouriteEvent}
def find_user_favourite_event_by_id(event_id):
return UserFavouriteEvent.query.filter_by(deleted_at=None,
user=current_user,
event_id=event_id).first()