forked from melli-labs/python-challenge
-
Notifications
You must be signed in to change notification settings - Fork 0
/
emilia.py
234 lines (173 loc) Β· 6.04 KB
/
emilia.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
223
224
225
226
227
228
229
230
231
232
233
234
from fastapi import FastAPI
app = FastAPI(
title="Emilia Hiring Challenge π©βπ»",
description="Help Emilia π© to fix our tests and get a job interview πΌποΈ!",
)
"""
Task 1 - Warmup
"""
@app.get("/task1/greet/{name}", tags=["Task 1"], summary="ππ©πͺπ¬π§πͺπΈ")
async def task1_greet(name: str) -> str:
"""Greet somebody in German, English or Spanish!"""
# Write your code below
...
return f"Hello {name}, I am Emilia."
"""
Task 2 - snake_case to cameCase
"""
from typing import Any
def camelize(key: str):
"""Takes string in snake_case format returns camelCase formatted version."""
# Write your code below
...
return key
@app.post("/task2/camelize", tags=["Task 2"], summary="πβ‘οΈπͺ")
async def task2_camelize(data: dict[str, Any]) -> dict[str, Any]:
"""Takes a JSON object and transfroms all keys from snake_case to camelCase."""
return {camelize(key): value for key, value in data.items()}
"""
Task 3 - Handle User Actions
"""
from pydantic import BaseModel
friends = {
"Matthias": ["Sahar", "Franziska", "Hans"],
"Stefan": ["Felix", "Ben", "Philip"],
}
class ActionRequest(BaseModel):
username: str
action: str
class ActionResponse(BaseModel):
message: str
def handle_call_action(action: str):
# Write your code below
...
return "π€ Why don't you call them yourself!"
def handle_reminder_action(action: str):
# Write your code below
...
return "π I can't even remember my own stuff!"
def handle_timer_action(action: str):
# Write your code below
...
return "β° I don't know how to read the clock!"
def handle_unknown_action(action: str):
# Write your code below
...
return "π€¬ #$!@"
@app.post("/task3/action", tags=["Task 3"], summary="π€")
def task3_action(request: ActionRequest):
"""Accepts an action request, recognizes its intent and forwards it to the corresponding action handler."""
# tip: you have to use the response model above and also might change the signature
# of the action handlers
# Write your code below
...
from random import choice
# There must be a better way!
handler = choice(
[
handle_call_action,
handle_reminder_action,
handle_timer_action,
handle_unknown_action,
]
)
return handler(request.action)
"""
Task 4 - Security
"""
from datetime import datetime, timedelta
from functools import partial
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
# create secret key with: openssl rand -hex 32
SECRET_KEY = "069d49a9c669ddc08f496352166b7b5d270ff64d3009fc297689aa8b0fb66d98"
ALOGRITHM = "HS256"
encode_jwt = partial(jwt.encode, key=SECRET_KEY, algorithm=ALOGRITHM)
decode_jwt = partial(jwt.decode, key=SECRET_KEY, algorithms=[ALOGRITHM])
_crypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
verify_password = _crypt_context.verify
hash_password = _crypt_context.hash
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/task4/token")
fake_users_db = {
"stefan": {
"username": "stefan",
"email": "[email protected]",
"hashed_password": hash_password("decent-espresso-by-john-buckmann"),
"secret": "I love pressure-profiled espresso β!",
},
"felix": {
"username": "felix",
"email": "[email protected]",
"hashed_password": hash_password("elm>javascript"),
"secret": "Rust π¦ is the best programming language ever!",
},
}
class User(BaseModel):
username: str
email: str
hashed_password: str
secret: str
class Token(BaseModel):
access_token: str
token_type: str
@app.post("/task4/token", response_model=Token, summary="π", tags=["Task 4"])
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
"""Allows registered users to obtain a bearer token."""
# fixme π¨, at the moment we allow everybody to obtain a token
# this is probably not very secure π‘οΈ ...
# tip: check the verify_password above
# Write your code below
...
payload = {
"sub": form_data.username,
"exp": datetime.utcnow() + timedelta(minutes=30),
}
return {
"access_token": encode_jwt(payload),
"token_type": "bearer",
}
def get_user(username: str) -> Optional[User]:
if username not in fake_users_db:
return
return User(**fake_users_db[username])
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
# check if the token πͺ is valid and return a user as specified by the tokens payload
# otherwise raise the credentials_exception above
# Write your code below
...
@app.get("/task4/users/{username}/secret", summary="π€«", tags=["Task 4"])
async def read_user_secret(
username: str, current_user: User = Depends(get_current_user)
):
"""Read a user's secret."""
# uppps π€ maybe we should check if the requested secret actually belongs to the user
# Write your code below
...
if user := get_user(username):
return user.secret
"""
Task and Help Routes
"""
from functools import partial
from pathlib import Path
from tomlkit.api import parse
messages = parse((Path(__file__).parent / "messages.toml").read_text("utf-8"))
@app.get("/", summary="π", tags=["Emilia"])
async def hello():
return messages["hello"]
identity = lambda x: x
for i in 1, 2, 3, 4:
task = messages[f"task{i}"]
info = partial(identity, task["info"])
help_ = partial(identity, task["help"])
tags = [f"Task {i}"]
app.get(f"/task{i}", summary="π", description=info(), tags=tags)(info)
app.get(f"/task{i}/help", summary="π", description=help_(), tags=tags)(help_)