-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathqueries.py
173 lines (132 loc) · 4.08 KB
/
queries.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
"""
Query Resolvers
"""
import json
import os
from typing import List
import requests
import strawberry
from db import ccdb, docsstoragedb
from models import CCRecruitment, StorageFile
# import all models and types
from otypes import (
CCRecruitmentType,
Info,
SignedURL,
SignedURLInput,
StorageFileType,
)
inter_communication_secret = os.getenv("INTER_COMMUNICATION_SECRET")
# fetch signed url from the files service
@strawberry.field
def signedUploadURL(details: SignedURLInput, info: Info) -> SignedURL:
"""
Uploads file to the files service by any user.
Args:
details (SignedURLInput): contains the details of the file to be
uploaded
info (Info): contains the user's context information.
Returns:
(SignedURL): A signed URL for uploading a file to the files service.
Raises:
Exception: Not logged in!
Exception: If the request failed.
"""
user = info.context.user
if not user:
raise Exception("Not logged in!")
# make request to files api
response = requests.get(
"http://files/signed-url",
params={
"user": json.dumps(user),
"static_file": "true" if details.static_file else "false",
"filename": details.filename,
"inter_communication_secret": inter_communication_secret,
"max_sizeMB": details.max_size_mb,
},
)
# error handling
if response.status_code != 200:
raise Exception(response.text)
return SignedURL(url=response.text)
@strawberry.field
def ccApplications(info: Info) -> List[CCRecruitmentType]:
"""
Returns list of all CC Applications for CC.
Args:
info (Info): contains the user's context information.
Returns:
(List[CCRecruitmentType]): A list of all CC Applications.
Raises:
Exception: Not logged in!
Exception: Not Authenticated to access this API!!
"""
user = info.context.user
if not user:
raise Exception("Not logged in!")
if user.get("role", None) not in ["cc"]:
raise Exception("Not Authenticated to access this API!!")
results = ccdb.find()
applications = [
CCRecruitmentType.from_pydantic(CCRecruitment.model_validate(result))
for result in results
]
return applications
@strawberry.field
def haveAppliedForCC(info: Info) -> bool:
"""
Finds whether any logged in user has applied for CC.
Args:
info (Info): contains the user's context information.
Returns:
(bool): True if the user has applied for CC, False otherwise.
Raises:
Exception: Not logged in!
Exception: Not Authenticated to access this API!!
"""
user = info.context.user
if not user:
raise Exception("Not logged in!")
if user.get("role", None) not in ["public"]:
raise Exception("Not Authenticated to access this API!!")
result = ccdb.find_one({"uid": user["uid"]})
if result:
return True
return False
# Storagefile queries
@strawberry.field
def storagefiles(filetype: str) -> List[StorageFileType]:
"""
Gets all the storage files, has public access
Args:
filetype (str): The type of file to get
Returns:
(List[StorageFileType]): A list of all storage files of the given type
"""
storage_files = docsstoragedb.find({"filetype": filetype})
return [
StorageFileType.from_pydantic(StorageFile.model_validate(storage_file))
for storage_file in storage_files
]
@strawberry.field
def storagefile(file_id: str) -> StorageFileType:
"""
Gets a single storage file by id, has public access
Args:
file_id (str): The id of the file to get
Returns:
(StorageFileType): The storage file with the given id
"""
storage_file = docsstoragedb.find_one({"_id": file_id})
return StorageFileType.from_pydantic(
StorageFile.model_validate(storage_file)
)
# register all queries
queries = [
signedUploadURL,
ccApplications,
haveAppliedForCC,
storagefiles,
storagefile,
]